From 2b2a36ae56e5bedc6a656e2cd0af71a1b2a20f36 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Tue, 28 Jul 2026 11:28:29 +0200 Subject: [PATCH] fix(sync): reject uploads behind state replacements (#9330) Persist the latest SYNC_IMPORT/BACKUP_IMPORT boundary and reject incremental uploads whose cursor has not observed it. Reconcile upgrade rows lazily, preserve explicit account-reset recovery, and cover cache, quota, PostgreSQL, and encrypted-client paths. --- ARCHITECTURE-DECISIONS.md | 5 + .../operation-log-architecture.md | 10 +- docs/wiki/3.01-API.md | 4 +- ...persync-encryption-password-change.spec.ts | 145 ++++++++++ .../super-sync-server/docs/architecture.md | 17 +- .../migration.sql | 5 + .../super-sync-server/prisma/schema.prisma | 1 + .../src/sync/sync.routes.ops-handler.ts | 112 ++++++-- .../src/sync/sync.service.ts | 132 +++++++-- .../super-sync-server/src/sync/sync.types.ts | 15 ++ ...an-slate-atomicity-sql.integration.spec.ts | 91 +++++++ .../tests/migration-sql.spec.ts | 11 + .../tests/sync-compressed-body.routes.spec.ts | 250 +++++++++++++++++- .../tests/sync.service.spec.ts | 126 ++++++++- 14 files changed, 870 insertions(+), 54 deletions(-) create mode 100644 packages/super-sync-server/prisma/migrations/20260727000000_add_state_replacement_seq/migration.sql diff --git a/ARCHITECTURE-DECISIONS.md b/ARCHITECTURE-DECISIONS.md index 8ed0dfc014..f0bd56a5a4 100644 --- a/ARCHITECTURE-DECISIONS.md +++ b/ARCHITECTURE-DECISIONS.md @@ -150,6 +150,10 @@ from PostgreSQL RepeatableRead snapshot isolation alone. - `REPAIR` uploads persist `repairBaseServerSeq` on the operation row. The HTTP handler rejects an obviously stale base before quota cleanup, and the upload transaction repeats the check under `SELECT ... FOR UPDATE` before insertion +- Regular uploads carrying `lastKnownServerSeq` use the same per-user row lock + to reject a batch behind the latest `SYNC_IMPORT` or `BACKUP_IMPORT` before + insertion. The durable replacement marker is reconciled lazily from retained + operations for rows created before the marker existed. - Markerless legacy repairs are compatibility records, not causal boundaries: they cannot drive download fast-forward, snapshot trust, history pruning, or server-generated restore points; snapshot replay across one fails closed @@ -172,6 +176,7 @@ from PostgreSQL RepeatableRead snapshot isolation alone. - Changing server sequence assignment - Changing transaction isolation for upload operations - Changing repair base-cursor validation or full-state history pruning +- Changing the state-replacement upload fence - Introducing multi-writer or multi-region upload processing --- diff --git a/docs/sync-and-op-log/operation-log-architecture.md b/docs/sync-and-op-log/operation-log-architecture.md index 1a2eee9da3..67cd767e61 100644 --- a/docs/sync-and-op-log/operation-log-architecture.md +++ b/docs/sync-and-op-log/operation-log-architecture.md @@ -988,7 +988,7 @@ The executable owner is with causal classification shared through [`classifyOpAgainstSyncImport`](../../packages/sync-core/src/sync-import-filter.ts). -### The Problem +### The Problem Without a Transport Fence Consider this scenario: @@ -1023,6 +1023,14 @@ suffix work remains valid. All of these decisions use vector clocks because the question is causal knowledge of the full-state boundary, not wall-clock recency. +SuperSync also prevents the invalid suffix from entering the server log. An +incremental upload carrying `lastKnownServerSeq` is serialized with state +replacement uploads on the user's sequence row. If its cursor predates the +latest `SYNC_IMPORT` or `BACKUP_IMPORT`, the whole batch is rejected and the +replacement is piggybacked before the client retries. Client-side filtering +remains necessary for other providers, retained legacy data, and defense in +depth. + See the field guide's [causality and conflict policy](./sync-architecture.html#causality) for the visual overview. diff --git a/docs/wiki/3.01-API.md b/docs/wiki/3.01-API.md index 81db871dc7..d10b5b816c 100755 --- a/docs/wiki/3.01-API.md +++ b/docs/wiki/3.01-API.md @@ -51,7 +51,7 @@ All require authentication. **Operations:** -- `POST /api/sync/ops` — Upload operations (incremental). Request: `ops[]`, `clientId`, optional `lastKnownServerSeq`, optional `requestId` (deduplication). +- `POST /api/sync/ops` — Upload operations (incremental). Request: `ops[]`, `clientId`, optional `lastKnownServerSeq`, optional `requestId` (deduplication). When `lastKnownServerSeq` is older than the latest state-replacing `SYNC_IMPORT` or `BACKUP_IMPORT`, the server rejects the batch and piggybacks that replacement so the client can apply it before retrying. - `GET /api/sync/ops?sinceSeq={seq}&limit={limit}&excludeClient={clientId}` — Download operations. Query: `sinceSeq` (required), `limit` (default 500, max 1000), `excludeClient` (optional). **Snapshots:** @@ -80,7 +80,7 @@ Exact schemas (Zod on server, TypeScript types in repo) are in `packages/super-s ### Error Codes -Structured error codes (for client handling) include: `VALIDATION_FAILED`, `INVALID_OP_ID`, `INVALID_OP_TYPE`, `INVALID_ENTITY_TYPE`, `INVALID_ENTITY_ID`, `INVALID_PAYLOAD`, `PAYLOAD_TOO_LARGE`, `INVALID_VECTOR_CLOCK`, `INVALID_CLIENT_ID`, `CONFLICT_CONCURRENT`, `CONFLICT_SUPERSEDED`, `REPAIR_STALE`, `DUPLICATE_OPERATION`, `RATE_LIMITED`, `STORAGE_QUOTA_EXCEEDED`, `ENCRYPTED_OPS_NOT_SUPPORTED`, `SYNC_IMPORT_EXISTS`, `INTERNAL_ERROR`. `REPAIR_STALE` means the repair snapshot did not include the server's current operation prefix; clients must download that suffix and rebuild the repair before retrying. Defined in `sync.types.ts` (`SYNC_ERROR_CODES`). +Structured error codes (for client handling) include: `VALIDATION_FAILED`, `INVALID_OP_ID`, `INVALID_OP_TYPE`, `INVALID_ENTITY_TYPE`, `INVALID_ENTITY_ID`, `INVALID_PAYLOAD`, `PAYLOAD_TOO_LARGE`, `INVALID_VECTOR_CLOCK`, `INVALID_CLIENT_ID`, `CONFLICT_CONCURRENT`, `CONFLICT_SUPERSEDED`, `REPAIR_STALE`, `DUPLICATE_OPERATION`, `RATE_LIMITED`, `STORAGE_QUOTA_EXCEEDED`, `ENCRYPTED_OPS_NOT_SUPPORTED`, `SYNC_IMPORT_EXISTS`, `INTERNAL_ERROR`. `REPAIR_STALE` means the repair snapshot did not include the server's current operation prefix; clients must download that suffix and rebuild the repair before retrying. A stale incremental upload behind a state replacement receives per-operation `INTERNAL_ERROR` results and the replacement as piggybacked operations; these results are retryable after applying the replacement. Defined in `sync.types.ts` (`SYNC_ERROR_CODES`). ### Rate Limits (per endpoint) diff --git a/e2e/tests/sync/supersync-encryption-password-change.spec.ts b/e2e/tests/sync/supersync-encryption-password-change.spec.ts index 5d9aaa7e0c..35873f258d 100644 --- a/e2e/tests/sync/supersync-encryption-password-change.spec.ts +++ b/e2e/tests/sync/supersync-encryption-password-change.spec.ts @@ -4,6 +4,7 @@ import { getSuperSyncConfig, createSimulatedClient, closeClient, + renameTask, waitForTask, type SimulatedE2EClient, } from '../../utils/supersync-helpers'; @@ -11,6 +12,11 @@ import { expectTaskNotVisible, expectTaskVisible, } from '../../utils/supersync-assertions'; +import { + SuperSyncDownloadOpsResponseSchema, + SuperSyncUploadOpsResponseSchema, + type SuperSyncServerOperation, +} from '../../../packages/shared-schema/src'; interface SyncStatusResponse { storageUsedBytes: number; @@ -48,6 +54,25 @@ const getSyncStatus = async ( return body; }; +const getServerOperations = async ( + baseUrl: string, + accessToken: string, +): Promise => { + const response = await fetch(`${baseUrl}/api/sync/ops?sinceSeq=0&limit=1000`, { + headers: { + Authorization: `Bearer ${accessToken}`, + }, + }); + + if (!response.ok) { + throw new Error( + `Failed to fetch server operations: ${response.status} ${await response.text()}`, + ); + } + + return SuperSyncDownloadOpsResponseSchema.parse(await response.json()).ops; +}; + const useServerDataIfPrompted = async (client: SimulatedE2EClient): Promise => { const appeared = await client.sync.syncImportConflictDialog .waitFor({ state: 'visible', timeout: 15000 }) @@ -297,6 +322,126 @@ test.describe('@supersync SuperSync Encryption Password Change', () => { } }); + test('#8730 rejects an old-key immediate upload after another client changes the password', async ({ + browser, + baseURL, + testRunId, + }, testInfo) => { + testInfo.setTimeout(240000); + const appUrl = baseURL || 'http://localhost:4242'; + let clientA: SimulatedE2EClient | null = null; + let clientB: SimulatedE2EClient | null = null; + + try { + const user = await createTestUser(testRunId); + const baseConfig = getSuperSyncConfig(user); + const oldPassword = `oldpass-8730-${testRunId}`; + const newPassword = `newpass-8730-${testRunId}`; + const originalTaskName = `Issue8730-Before-${testRunId}`; + const staleTaskName = `Issue8730-Stale-${testRunId}`; + + clientA = await createSimulatedClient(browser, appUrl, 'A', testRunId); + await clientA.sync.setupSuperSync({ + ...baseConfig, + isEncryptionEnabled: true, + password: oldPassword, + }); + + clientB = await createSimulatedClient(browser, appUrl, 'B', testRunId); + await clientB.sync.setupSuperSync({ + ...baseConfig, + isEncryptionEnabled: true, + password: oldPassword, + }); + + await clientA.workView.addTask(originalTaskName); + await clientA.sync.syncAndWait(); + await clientB.sync.syncAndWait(); + await waitForTask(clientB.page, originalTaskName); + + await clientA.sync.changeEncryptionPassword(newPassword); + + const clientBDecryptDialog = clientB.page.locator('dialog-handle-decrypt-error'); + await clientB.sync.syncBtn.click(); + await expect(clientBDecryptDialog).toBeVisible({ timeout: 30000 }); + await clientBDecryptDialog.getByRole('button', { name: /cancel/i }).click(); + await expect(clientBDecryptDialog).toBeHidden(); + await clientB.sync.syncSpinner.waitFor({ state: 'hidden', timeout: 30000 }); + + await clientB.page.evaluate(() => { + const testGlobal = globalThis as typeof globalThis & { + __SP_E2E_BLOCK_AUTO_SYNC?: boolean; + __SP_E2E_BLOCK_IMMEDIATE_UPLOAD?: boolean; + }; + testGlobal.__SP_E2E_BLOCK_AUTO_SYNC = true; + testGlobal.__SP_E2E_BLOCK_IMMEDIATE_UPLOAD = false; + }); + + const staleUploadResponsePromise = clientB.page.waitForResponse( + (response) => + response.request().method() === 'POST' && + new URL(response.url()).pathname === '/api/sync/ops', + { timeout: 30000 }, + ); + await renameTask(clientB, originalTaskName, staleTaskName); + const staleUploadResponse = await staleUploadResponsePromise; + expect(staleUploadResponse.ok()).toBe(true); + + const uploadResponseBody = SuperSyncUploadOpsResponseSchema.parse( + await staleUploadResponse.json(), + ); + expect(uploadResponseBody.results).toHaveLength(1); + const staleUploadResult = uploadResponseBody.results[0]; + const serverOperations = await getServerOperations( + baseConfig.baseUrl, + baseConfig.accessToken, + ); + const passwordChangeOp = serverOperations.find( + ({ op }) => + op.opType === 'SYNC_IMPORT' && op.syncImportReason === 'PASSWORD_CHANGED', + ); + expect( + passwordChangeOp, + 'the password change must leave a real encrypted full-state operation', + ).toBeDefined(); + if (!passwordChangeOp) { + throw new Error('The server did not return the password-change SYNC_IMPORT'); + } + expect(passwordChangeOp.op.isPayloadEncrypted).toBe(true); + expect(typeof passwordChangeOp.op.payload).toBe('string'); + expect(uploadResponseBody).toEqual( + expect.objectContaining({ + newOps: expect.arrayContaining([ + expect.objectContaining({ + op: expect.objectContaining({ id: passwordChangeOp.op.id }), + }), + ]), + }), + ); + + const serverStaleTaskOp = serverOperations.find( + ({ op }) => op.id === staleUploadResult.opId, + ); + const clientADecryptDialog = clientA.page.locator('dialog-handle-decrypt-error'); + + expect( + staleUploadResult.accepted, + `stale old-key upload must be rejected after the PASSWORD_CHANGED clean slate` + + (staleUploadResult.errorCode + ? ` (server error code: ${staleUploadResult.errorCode})` + : ''), + ).toBe(false); + expect(staleUploadResult.errorCode).toBe('INTERNAL_ERROR'); + expect(serverStaleTaskOp).toBeUndefined(); + await clientA.sync.syncAndWait(); + await expect(clientADecryptDialog).toBeHidden(); + expect(await clientA.sync.hasSyncError()).toBe(false); + } finally { + if (clientA) await closeClient(clientA); + if (clientB) await closeClient(clientB); + } + }); + test('Incompatible stale client adopts Client A clean slate and drops pending data', async ({ browser, baseURL, diff --git a/packages/super-sync-server/docs/architecture.md b/packages/super-sync-server/docs/architecture.md index 1c2a86f1d6..2112e0faf0 100644 --- a/packages/super-sync-server/docs/architecture.md +++ b/packages/super-sync-server/docs/architecture.md @@ -77,6 +77,13 @@ commit conflicting operations. A causal `REPAIR` additionally locks that row and must prove `repairBaseServerSeq === lastSeq`. Incoming vector clocks are compared before being pruned for storage. +A regular upload carrying `lastKnownServerSeq` uses the same row lock to compare +its cursor with `user_sync_state.latestStateReplacementSeq`. A cursor behind +the latest `SYNC_IMPORT` or `BACKUP_IMPORT` is rejected before any operation is +inserted, and the replacement is piggybacked without excluding its author. The +nullable marker is reconciled lazily from retained operations after an upgrade; +zero records that the reconciliation found no replacement. + A clean-slate full-state upload deletes the prior dataset but preserves `lastSeq`, preventing sequence reuse visible to existing clients. Only explicit `DELETE /api/sync/data` erases the entire dataset and resets the sequence to @@ -92,11 +99,11 @@ This serialization mechanism is a load-bearing decision; see - `operations` is append-on-write, not retained forever. Rows are immutable while retained; cleanup, quota recovery, clean-slate replacement, and explicit data deletion can remove them. -- `user_sync_state` owns `lastSeq`, the optional compressed snapshot cache, and - the latest causal full-state marker. `sync_devices` is used only for per-device - identity/metadata and last-seen tracking. Its `lastAckedSeq` field is dormant - legacy schema state: current sync and retention code neither advances nor - reads it. +- `user_sync_state` owns `lastSeq`, the optional compressed snapshot cache, the + latest causal full-state marker, and the latest explicit state-replacement + boundary. `sync_devices` is used only for per-device identity/metadata and + last-seen tracking. Its `lastAckedSeq` field is dormant legacy schema state: + current sync and retention code neither advances nor reads it. - Normal sync bootstraps from operation rows. `GET /ops` can fast-forward to the latest causal full-state operation; clients do not download the server's cached snapshot blob. diff --git a/packages/super-sync-server/prisma/migrations/20260727000000_add_state_replacement_seq/migration.sql b/packages/super-sync-server/prisma/migrations/20260727000000_add_state_replacement_seq/migration.sql new file mode 100644 index 0000000000..1af22ea0ca --- /dev/null +++ b/packages/super-sync-server/prisma/migrations/20260727000000_add_state_replacement_seq/migration.sql @@ -0,0 +1,5 @@ +-- Persist the latest state-replacing boundary independently from REPAIR +-- snapshots. Incremental uploads use this cursor to prove they have downloaded +-- the latest SYNC_IMPORT/BACKUP_IMPORT before writing. +ALTER TABLE "user_sync_state" + ADD COLUMN "latest_state_replacement_seq" INTEGER; diff --git a/packages/super-sync-server/prisma/schema.prisma b/packages/super-sync-server/prisma/schema.prisma index e7d69faf55..fb8809e340 100644 --- a/packages/super-sync-server/prisma/schema.prisma +++ b/packages/super-sync-server/prisma/schema.prisma @@ -134,6 +134,7 @@ model UserSyncState { snapshotSchemaVersion Int? @default(1) @map("snapshot_schema_version") latestFullStateSeq Int? @map("latest_full_state_seq") latestFullStateVectorClock Json? @map("latest_full_state_vector_clock") + latestStateReplacementSeq Int? @map("latest_state_replacement_seq") user User @relation(fields: [userId], references: [id], onDelete: Cascade) 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 feda3e9be7..b40885081a 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 @@ -5,8 +5,10 @@ import { Logger } from '../logger'; import { getSyncService } from './sync.service'; import { getWsConnectionService } from './services/websocket-connection.service'; import { + createStateReplacementRequiredResults, Operation, ServerOperation, + STATE_REPLACEMENT_REQUIRED_ERROR, SYNC_ERROR_CODES, UploadOpsRequest, UploadOpsResponse, @@ -32,6 +34,15 @@ import { } from './sync.routes.quota'; import { createOpsRequestFingerprint } from './services/request-deduplication.service'; +const isStateReplacementFenceRejection = (results: UploadResult[]): boolean => + results.length > 0 && + results.every( + (result) => + !result.accepted && + result.errorCode === SYNC_ERROR_CODES.INTERNAL_ERROR && + result.error === STATE_REPLACEMENT_REQUIRED_ERROR, + ); + export const uploadOpsHandler = async ( req: FastifyRequest<{ Body: UploadOpsRequest }>, reply: FastifyReply, @@ -129,28 +140,46 @@ export const uploadOpsHandler = async ( requestId, () => requestFingerprint, ); - if (cachedResults) { - Logger.info(`[user:${userId}] Returning cached results for request ${requestId}`); - + if (cachedResults && lastKnownServerSeq !== undefined) { // IMPORTANT: Recompute piggybacked ops using the retry request's lastKnownServerSeq. // 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: ServerOperation[] | undefined; - let latestSeq: number; - let hasMorePiggyback = false; - const PIGGYBACK_LIMIT = 500; - - if (lastKnownServerSeq !== undefined) { - const opsResult = await syncService.getOpsSinceWithSeq( - userId, - lastKnownServerSeq, - clientId, - PIGGYBACK_LIMIT, - false, + const opsResult = await syncService.getOpsSinceWithSeq( + userId, + lastKnownServerSeq, + clientId, + 500, + false, + ); + const latestStateReplacementSeq = + await syncService.getLatestStateReplacementSeq(userId); + const cachedResultPredatesStateReplacement = + latestStateReplacementSeq !== null && + (lastKnownServerSeq < latestStateReplacementSeq || + cachedResults.some( + (result) => + result.accepted && + (result.serverSeq === undefined || + result.serverSeq < latestStateReplacementSeq), + )); + if (cachedResultPredatesStateReplacement) { + // The piggyback read comes first, so a replacement committed before + // or during it is visible to this boundary read. A replacement that + // commits later is beyond opsResult.latestSeq and will be downloaded + // on the next sync. + Logger.info( + `[user:${userId}] Ignoring cached upload result from before the latest state replacement`, ); - newOps = opsResult.ops; - latestSeq = opsResult.latestSeq; + } else { + Logger.info( + `[user:${userId}] Returning cached results for request ${requestId}`, + ); + + const newOps = opsResult.ops; + const latestSeq = opsResult.latestSeq; + let hasMorePiggyback = false; + const PIGGYBACK_LIMIT = 500; // Check if there are more ops beyond what we piggybacked if (newOps.length === PIGGYBACK_LIMIT) { @@ -164,16 +193,24 @@ export const uploadOpsHandler = async ( (hasMorePiggyback ? ` (has more)` : ''), ); } - } else { - latestSeq = await syncService.getLatestSeq(userId); + + return reply.send({ + results: cachedResults, + newOps: newOps.length ? newOps : undefined, + latestSeq, + deduplicated: true, + ...(hasMorePiggyback ? { hasMorePiggyback: true } : {}), + } as UploadOpsResponse & { deduplicated: boolean }); } + } else if (cachedResults) { + Logger.info(`[user:${userId}] Returning cached results for request ${requestId}`); + + const latestSeq = await syncService.getLatestSeq(userId); return reply.send({ results: cachedResults, - newOps: newOps?.length ? newOps : undefined, latestSeq, deduplicated: true, - ...(hasMorePiggyback ? { hasMorePiggyback: true } : {}), } as UploadOpsResponse & { deduplicated: boolean }); } } @@ -206,7 +243,20 @@ export const uploadOpsHandler = async ( `charged at APPROX_BYTES_PER_OP for user=${userId} (gate)`, ); } - const quotaOk = await enforceStorageQuota(userId, estimatedDelta, reply); + const initialQuota = await syncService.checkStorageQuota(userId, estimatedDelta); + if (!initialQuota.allowed && lastKnownServerSeq !== undefined) { + const latestStateReplacementSeq = + await syncService.getLatestStateReplacementSeq(userId); + if ( + latestStateReplacementSeq !== null && + lastKnownServerSeq < latestStateReplacementSeq + ) { + return createStateReplacementRequiredResults(typedOpsForGate); + } + } + const quotaOk = + initialQuota.allowed || + (await enforceStorageQuota(userId, estimatedDelta, reply)); if (!quotaOk) return null; // Process operations - cast to Operation[] since Zod validates the structure. @@ -220,20 +270,21 @@ export const uploadOpsHandler = async ( ops as unknown as Operation[], undefined, requestStartOccupiedIds, + undefined, + false, + lastKnownServerSeq, ); return uploadResults; }, ); if (!results) return; + const requiresStateReplacementDownload = isStateReplacementFenceRejection(results); // Cache results for deduplication if requestId was provided. - // Skip caching when the whole batch rolled back (INTERNAL_ERROR): nothing - // was committed, so the deterministic-requestId retry must re-process - // rather than be served the cached transient failure for the dedup TTL - // (REQUEST_DEDUP_TTL_MS = 5 min). INTERNAL_ERROR is produced ONLY by - // uploadOps' rollback catch (sync.service.ts), which maps the *entire* - // batch to it, so a single match means the transaction failed. #8332 + // Skip caching transient INTERNAL_ERROR results: neither transaction + // rollbacks nor state-replacement fence rejections committed this batch, + // so deterministic-requestId retries must re-process. #8332 if ( requestId && requestFingerprint && @@ -266,7 +317,10 @@ export const uploadOpsHandler = async ( const opsResult = await syncService.getOpsSinceWithSeq( userId, lastKnownServerSeq, - clientId, + // The replacement may share this client ID (for example a migration + // snapshot). A fenced retry must receive the boundary before its local + // cursor can advance, even though ordinary piggybacking excludes self. + requiresStateReplacementDownload ? undefined : clientId, PIGGYBACK_LIMIT, false, ); diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index a1f4cb25df..6605c9466e 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -7,6 +7,7 @@ import { DEFAULT_SYNC_CONFIG, VectorClock, SYNC_ERROR_CODES, + createStateReplacementRequiredResults, } from './sync.types'; import { Logger } from '../logger'; import { Prisma } from '@prisma/client'; @@ -141,6 +142,7 @@ export class SyncService { requestStartOccupiedIds?: ReadonlySet, repairBaseServerSeq?: number, allowLegacyRepairWithoutBase: boolean = false, + lastKnownServerSeq?: number, ): Promise { if (isCleanSlate && ops.length === 0) { return []; @@ -198,22 +200,74 @@ export class SyncService { // Use transaction to acquire write lock and ensure atomicity await prisma.$transaction( async (tx) => { - if (containsRepair && !isLegacyRepairUpload) { - // Serialize the causal precondition and the later insert through the - // same per-user sequence row. A concurrent upload either commits - // first (making this base stale) or waits and lands after REPAIR. + const needsSyncStateLock = + shouldCleanSlate || + lastKnownServerSeq !== undefined || + (containsRepair && !isLegacyRepairUpload); + let currentServerSeq = 0; + if (needsSyncStateLock) { + // Serialize state replacements, cursor checks, and later inserts on + // the same per-user row. Whichever request acquires this lock first + // defines the safe order seen by every other server instance. await tx.userSyncState.upsert({ where: { userId }, create: { userId, lastSeq: 0 }, update: {}, }); - const rows = await tx.$queryRaw>` - SELECT last_seq AS "lastSeq" + const rows = await tx.$queryRaw< + Array<{ + lastSeq: number; + latestStateReplacementSeq: number | null; + }> + >` + SELECT + last_seq AS "lastSeq", + latest_state_replacement_seq AS "latestStateReplacementSeq" FROM user_sync_state WHERE user_id = ${userId} FOR UPDATE `; - const currentServerSeq = rows[0]?.lastSeq ?? 0; + currentServerSeq = rows[0]?.lastSeq ?? 0; + let latestStateReplacementSeq = rows[0]?.latestStateReplacementSeq ?? null; + if (latestStateReplacementSeq === null) { + // The column is intentionally not backfilled during migration: + // the supported Compose deploy keeps the old process serving + // while migrations run. Resolve retained replacements lazily + // after the new process owns the write path, then persist the + // answer for subsequent uploads. + const retainedReplacement = await tx.operation.findFirst({ + where: { + userId, + opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT'] }, + }, + orderBy: { serverSeq: 'desc' }, + select: { serverSeq: true }, + }); + // Zero is a resolved "no retained replacement" sentinel. Keeping + // null exclusively for unresolved upgrade rows avoids repeating + // this indexed lookup on every upload for ordinary accounts. + latestStateReplacementSeq = retainedReplacement?.serverSeq ?? 0; + await tx.userSyncState.update({ + where: { userId }, + data: { latestStateReplacementSeq }, + }); + uploadDbRoundtrips++; + } + if ( + lastKnownServerSeq !== undefined && + latestStateReplacementSeq !== null && + lastKnownServerSeq < latestStateReplacementSeq + ) { + Logger.warn( + `[user:${userId}] Rejecting upload from stale state replacement cursor ` + + `(client=${lastKnownServerSeq}, required=${latestStateReplacementSeq})`, + ); + results.push(...createStateReplacementRequiredResults(ops)); + return; + } + } + + if (containsRepair && !isLegacyRepairUpload) { if ( repairBaseServerSeq === undefined || repairBaseServerSeq !== currentServerSeq @@ -267,6 +321,7 @@ export class SyncService { snapshotAt: null, latestFullStateSeq: null, latestFullStateVectorClock: Prisma.DbNull, + latestStateReplacementSeq: null, }, }); @@ -307,12 +362,14 @@ export class SyncService { // We assume user exists in `users` table because of foreign key, // but if `uploadOps` is called, authentication should have verified user existence. // However, `user_sync_state` might not exist yet. - await tx.userSyncState.upsert({ - where: { userId }, - create: { userId, lastSeq: 0 }, - update: {}, // No-op update to ensure it exists - }); - uploadDbRoundtrips++; + if (!needsSyncStateLock) { + await tx.userSyncState.upsert({ + where: { userId }, + create: { userId, lastSeq: 0 }, + update: {}, // No-op update to ensure it exists + }); + uploadDbRoundtrips++; + } const firstOperationById = new Map< string, @@ -352,6 +409,32 @@ export class SyncService { } } } + + let latestAcceptedStateReplacementSeq: number | undefined; + for (let index = 0; index < ops.length; index++) { + const op = ops[index]; + const result = results[index]; + if ( + result?.accepted && + result.serverSeq !== undefined && + (op.opType === 'SYNC_IMPORT' || op.opType === 'BACKUP_IMPORT') + ) { + latestAcceptedStateReplacementSeq = Math.max( + latestAcceptedStateReplacementSeq ?? 0, + result.serverSeq, + ); + } + } + if (latestAcceptedStateReplacementSeq !== undefined) { + await tx.userSyncState.update({ + where: { userId }, + data: { + latestStateReplacementSeq: latestAcceptedStateReplacementSeq, + }, + }); + uploadDbRoundtrips++; + } + if (unserializableAccepted > 0) { Logger.warn( `computeOpsStorageBytes: ${unserializableAccepted} unserializable op(s) ` + @@ -639,6 +722,26 @@ export class SyncService { ); } + async getLatestStateReplacementSeq(userId: number): Promise { + const syncState = await prisma.userSyncState.findUnique({ + where: { userId }, + select: { latestStateReplacementSeq: true }, + }); + const latestStateReplacementSeq = syncState?.latestStateReplacementSeq; + if (typeof latestStateReplacementSeq === 'number') { + return latestStateReplacementSeq; + } + const retainedReplacement = await prisma.operation.findFirst({ + where: { + userId, + opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT'] }, + }, + orderBy: { serverSeq: 'desc' }, + select: { serverSeq: true }, + }); + return retainedReplacement?.serverSeq ?? null; + } + cacheOpsRequestResults( userId: number, requestId: string, @@ -772,8 +875,7 @@ export class SyncService { // Delete sync state entirely, resetting lastSeq to 0. // Unlike uploadOps clean slate (which preserves lastSeq), account reset // intentionally wipes everything. Clients detect the wipe via latestSeq=0 - // and trigger a full state re-upload. This is correct because account reset - // (e.g., encryption password change) requires ALL clients to re-sync. + // and trigger a full state re-upload. await tx.userSyncState.deleteMany({ where: { userId } }); // Reset storage usage diff --git a/packages/super-sync-server/src/sync/sync.types.ts b/packages/super-sync-server/src/sync/sync.types.ts index 64140a678e..c85e71f6ce 100644 --- a/packages/super-sync-server/src/sync/sync.types.ts +++ b/packages/super-sync-server/src/sync/sync.types.ts @@ -81,6 +81,9 @@ export const SYNC_ERROR_CODES = { INTERNAL_ERROR: 'INTERNAL_ERROR', } as const; +export const STATE_REPLACEMENT_REQUIRED_ERROR = + 'Download the latest full-state replacement before retrying'; + export type SyncErrorCode = (typeof SYNC_ERROR_CODES)[keyof typeof SYNC_ERROR_CODES]; export type ConflictType = @@ -304,6 +307,18 @@ export interface UploadResult { existingClock?: VectorClock; } +export const createStateReplacementRequiredResults = ( + ops: ReadonlyArray>, +): UploadResult[] => + ops.map((op) => ({ + opId: op.id, + accepted: false, + error: STATE_REPLACEMENT_REQUIRED_ERROR, + // Released clients already leave INTERNAL_ERROR operations pending and + // process piggybacked operations before retrying. + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + })); + /** * Internal return of the serial-path `processOperation`: the client-facing * `UploadResult` plus the op's storage size, computed once at the persist site, diff --git a/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts b/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts index 12d7c087c9..fed956cd60 100644 --- a/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts +++ b/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts @@ -89,6 +89,7 @@ describeWithDb('Clean-slate upload atomicity (PostgreSQL)', () => { snapshotAt: null, latestFullStateSeq: null, latestFullStateVectorClock: null, + latestStateReplacementSeq: null, }, }); await prisma.user.update({ @@ -139,4 +140,94 @@ describeWithDb('Clean-slate upload atomicity (PostgreSQL)', () => { ).toBeNull(); }, ); + + it.each([ + { name: 'serial upload path', batchUpload: false }, + { name: 'batch upload path', batchUpload: true }, + ] as const)( + 'rejects stale deltas after a replacement without blocking current clients ($name)', + async ({ batchUpload }) => { + const service = new SyncService({ batchUpload }); + const replacement = makeOp({ + id: `state-replacement-${batchUpload}`, + opType: 'SYNC_IMPORT', + actionType: '[SP_ALL] Load(import) all data', + entityType: 'ALL', + entityId: undefined, + payload: { task: { ids: ['replacement-task'] } }, + vectorClock: { [CLIENT_ID]: 1 }, + syncImportReason: 'FORCE_UPLOAD', + }); + const replacementResult = await service.uploadOps( + TEST_USER_ID, + CLIENT_ID, + [replacement], + true, + ); + const replacementSeq = replacementResult[0].serverSeq; + expect(replacementResult[0].accepted).toBe(true); + expect(replacementSeq).toBeDefined(); + if (replacementSeq === undefined) { + throw new Error('State replacement did not receive a server sequence'); + } + // Simulate an upgrade from a server version that wrote the retained + // replacement operation before the new boundary column was maintained. + await prisma.userSyncState.update({ + where: { userId: TEST_USER_ID }, + data: { latestStateReplacementSeq: null }, + }); + + const staleDelta = makeOp({ + id: `stale-delta-${batchUpload}`, + clientId: 'stale-client', + entityId: 'stale-task', + vectorClock: { 'stale-client': 1 }, + }); + const staleResult = await service.uploadOps( + TEST_USER_ID, + staleDelta.clientId, + [staleDelta], + undefined, + undefined, + undefined, + false, + replacementSeq - 1, + ); + + expect(staleResult[0]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }), + ); + expect( + await prisma.operation.findUnique({ where: { id: staleDelta.id } }), + ).toBeNull(); + + const currentDelta = makeOp({ + id: `current-delta-${batchUpload}`, + clientId: 'current-client', + entityId: 'current-task', + vectorClock: { 'current-client': 1 }, + }); + const currentResult = await service.uploadOps( + TEST_USER_ID, + currentDelta.clientId, + [currentDelta], + undefined, + undefined, + undefined, + false, + replacementSeq, + ); + + expect(currentResult[0].accepted).toBe(true); + expect( + await prisma.userSyncState.findUniqueOrThrow({ + where: { userId: TEST_USER_ID }, + select: { latestStateReplacementSeq: true }, + }), + ).toEqual({ latestStateReplacementSeq: replacementSeq }); + }, + ); }); diff --git a/packages/super-sync-server/tests/migration-sql.spec.ts b/packages/super-sync-server/tests/migration-sql.spec.ts index 775c29c739..ad9755fc5d 100644 --- a/packages/super-sync-server/tests/migration-sql.spec.ts +++ b/packages/super-sync-server/tests/migration-sql.spec.ts @@ -16,6 +16,17 @@ const allMigrationSql = (): string => .join('\n'); describe('performance migrations', () => { + it('adds the durable state-replacement boundary without a blocking backfill', () => { + const migrationSql = readMigration('20260727000000_add_state_replacement_seq'); + + expect(migrationSql).toMatch( + /ALTER TABLE "user_sync_state"\s+ADD COLUMN "latest_state_replacement_seq" INTEGER/i, + ); + expect(migrationSql).not.toMatch(/\bUPDATE\b/i); + expect(migrationSql).not.toMatch(/\bDROP\s+TABLE\b/i); + expect(migrationSql).not.toMatch(/\bBEGIN\b|\bCOMMIT\b/i); + }); + it('adds the entity sequence index without a blocking or destructive migration', () => { const migrationSql = readFileSync( join( 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 17179b5644..cd5a4a9260 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 @@ -8,6 +8,7 @@ const mocks = vi.hoisted(() => { const syncService = { isRateLimited: vi.fn(), checkOpsRequestDedup: vi.fn(), + getLatestStateReplacementSeq: vi.fn(), cacheOpsRequestResults: vi.fn(), checkSnapshotRequestDedup: vi.fn(), cacheSnapshotRequestResult: vi.fn(), @@ -68,7 +69,10 @@ vi.mock('../src/db', () => ({ })); import { syncRoutes } from '../src/sync/sync.routes'; -import { SYNC_ERROR_CODES } from '../src/sync/sync.types'; +import { + STATE_REPLACEMENT_REQUIRED_ERROR, + SYNC_ERROR_CODES, +} from '../src/sync/sync.types'; import { computeOpStorageBytes } from '../src/sync/sync.const'; import { CURRENT_SCHEMA_VERSION, SUPER_SYNC_MAX_OPS_PER_UPLOAD } from '@sp/shared-schema'; @@ -116,6 +120,7 @@ describe('Sync compressed body routes', () => { vi.clearAllMocks(); mocks.syncService.isRateLimited.mockReturnValue(false); mocks.syncService.checkOpsRequestDedup.mockReturnValue(null); + mocks.syncService.getLatestStateReplacementSeq.mockResolvedValue(null); mocks.syncService.checkSnapshotRequestDedup.mockReturnValue(null); mocks.syncService.getMaxClockDriftMs.mockReturnValue(60_000); mocks.syncService.filterValidOpsForQuota.mockImplementation((ops: unknown[]) => ops); @@ -260,6 +265,9 @@ describe('Sync compressed body routes', () => { [validOp, invalidOp], undefined, new Set(), + undefined, + false, + undefined, ); }); @@ -325,6 +333,9 @@ describe('Sync compressed body routes', () => { [validOp, invalidLargeOp], undefined, new Set(), + undefined, + false, + undefined, ); }); @@ -380,6 +391,9 @@ describe('Sync compressed body routes', () => { [duplicateOp, newOp], undefined, new Set([duplicateOp.id]), + undefined, + false, + undefined, ); }); @@ -430,6 +444,9 @@ describe('Sync compressed body routes', () => { [incomingOp], undefined, new Set([incomingOp.id]), + undefined, + false, + undefined, ); }); @@ -706,6 +723,237 @@ describe('Sync compressed body routes', () => { 500, false, ); + expect(mocks.syncService.uploadOps).toHaveBeenCalledWith( + 1, + clientId, + payload.ops, + undefined, + new Set(), + undefined, + false, + 0, + ); + }); + + it('reprocesses stale cached results and includes a same-client replacement', async () => { + const clientId = 'same-client-replacement'; + const staleResult = { + opId: 'op-1', + accepted: false, + error: STATE_REPLACEMENT_REQUIRED_ERROR, + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }; + const replacement = { + serverSeq: 3, + op: { + ...createOp(clientId), + id: 'state-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + syncImportReason: 'FORCE_UPLOAD', + }, + receivedAt: Date.now(), + }; + mocks.syncService.checkOpsRequestDedup.mockReturnValue([ + { opId: 'op-1', accepted: true, serverSeq: 2 }, + ]); + mocks.syncService.getLatestStateReplacementSeq.mockResolvedValue(3); + mocks.syncService.uploadOps.mockResolvedValue([staleResult]); + mocks.syncService.getOpsSinceWithSeq.mockResolvedValue({ + ops: [replacement], + latestSeq: 3, + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [createOp(clientId)], + clientId, + lastKnownServerSeq: 2, + requestId: 'pre-replacement-request', + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + results: [staleResult], + newOps: [replacement], + latestSeq: 3, + }); + expect(mocks.syncService.uploadOps).toHaveBeenCalledOnce(); + expect(mocks.syncService.getOpsSinceWithSeq).toHaveBeenCalledWith( + 1, + 2, + undefined, + 500, + false, + ); + }); + + it('reprocesses a cached success deleted by a replacement at the current cursor', async () => { + const clientId = 'current-after-replacement'; + mocks.syncService.checkOpsRequestDedup.mockReturnValue([ + { opId: 'op-1', accepted: true, serverSeq: 2 }, + ]); + mocks.syncService.getLatestStateReplacementSeq.mockResolvedValue(3); + mocks.syncService.uploadOps.mockResolvedValue([ + { opId: 'op-1', accepted: true, serverSeq: 4 }, + ]); + mocks.syncService.getOpsSinceWithSeq.mockResolvedValue({ + ops: [], + latestSeq: 4, + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [createOp(clientId)], + clientId, + lastKnownServerSeq: 3, + requestId: 'accepted-before-replacement', + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + results: [{ opId: 'op-1', accepted: true, serverSeq: 4 }], + latestSeq: 4, + }); + expect(mocks.syncService.uploadOps).toHaveBeenCalledOnce(); + expect(mocks.syncService.cacheOpsRequestResults).toHaveBeenCalledWith( + 1, + 'accepted-before-replacement', + [{ opId: 'op-1', accepted: true, serverSeq: 4 }], + expect.any(String), + ); + }); + + it('rechecks the replacement boundary after reading a cached retry response', async () => { + const clientId = 'replacement-race-client'; + const staleResult = { + opId: 'op-1', + accepted: false, + error: STATE_REPLACEMENT_REQUIRED_ERROR, + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }; + const replacement = { + serverSeq: 3, + op: { + ...createOp(clientId), + id: 'concurrent-state-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + syncImportReason: 'PASSWORD_CHANGED', + }, + receivedAt: Date.now(), + }; + mocks.syncService.checkOpsRequestDedup.mockReturnValue([ + { opId: 'op-1', accepted: true, serverSeq: 2 }, + ]); + mocks.syncService.getLatestStateReplacementSeq.mockResolvedValue(3); + mocks.syncService.getOpsSinceWithSeq + .mockResolvedValueOnce({ + ops: [], + latestSeq: 2, + }) + .mockResolvedValueOnce({ + ops: [replacement], + latestSeq: 3, + }); + mocks.syncService.uploadOps.mockResolvedValue([staleResult]); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [createOp(clientId)], + clientId, + lastKnownServerSeq: 2, + requestId: 'replacement-race', + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + results: [staleResult], + newOps: [replacement], + latestSeq: 3, + }); + expect(mocks.syncService.getLatestStateReplacementSeq).toHaveBeenCalledOnce(); + expect(mocks.syncService.uploadOps).toHaveBeenCalledOnce(); + }); + + it('returns the replacement fence before destructive quota cleanup', async () => { + const clientId = 'stale-near-quota'; + const replacement = { + serverSeq: 3, + op: { + ...createOp('replacement-client'), + id: 'state-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + syncImportReason: 'PASSWORD_CHANGED', + }, + receivedAt: Date.now(), + }; + mocks.syncService.checkStorageQuota.mockResolvedValue({ + allowed: false, + currentUsage: 100, + quota: 100, + }); + mocks.syncService.getLatestStateReplacementSeq.mockResolvedValue(3); + mocks.syncService.getOpsSinceWithSeq.mockResolvedValue({ + ops: [replacement], + latestSeq: 3, + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [createOp(clientId)], + clientId, + lastKnownServerSeq: 2, + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + results: [ + { + opId: 'op-1', + accepted: false, + error: STATE_REPLACEMENT_REQUIRED_ERROR, + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }, + ], + newOps: [replacement], + latestSeq: 3, + }); + expect(mocks.syncService.updateStorageUsage).not.toHaveBeenCalled(); + expect(mocks.syncService.freeStorageForUpload).not.toHaveBeenCalled(); + expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); }); it('should skip snapshot metadata for deduplicated retry piggyback downloads', async () => { diff --git a/packages/super-sync-server/tests/sync.service.spec.ts b/packages/super-sync-server/tests/sync.service.spec.ts index 48bf4a012f..0793a3a2a6 100644 --- a/packages/super-sync-server/tests/sync.service.spec.ts +++ b/packages/super-sync-server/tests/sync.service.spec.ts @@ -430,9 +430,11 @@ vi.mock('../src/db', async () => { } if (sql.includes('FROM user_sync_state') && sql.includes('FOR UPDATE')) { const [txUserId] = params as [number]; + const syncState = state.userSyncStates.get(txUserId); return [ { - lastSeq: state.userSyncStates.get(txUserId)?.lastSeq ?? 0, + lastSeq: syncState?.lastSeq ?? 0, + latestStateReplacementSeq: syncState?.latestStateReplacementSeq ?? null, }, ]; } @@ -913,6 +915,127 @@ describe('SyncService', () => { }); describe('uploadOps', () => { + it('rejects a cursor behind the latest state replacement but allows its boundary', async () => { + const service = new SyncService({ batchUpload: true }); + const op = makeOp({ id: 'post-replacement-edit' }); + const replacement = makeOp({ + id: 'retained-state-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + }); + testState.operations.set(replacement.id, { + ...replacement, + userId, + serverSeq: 3, + entityId: null, + entityIds: [], + payloadBytes: BigInt(1), + clientTimestamp: BigInt(replacement.timestamp), + receivedAt: BigInt(replacement.timestamp), + isPayloadEncrypted: false, + syncImportReason: null, + repairBaseServerSeq: null, + }); + testState.userSyncStates.set(userId, { + userId, + lastSeq: 4, + latestStateReplacementSeq: null, + }); + + const staleResults = await service.uploadOps( + userId, + clientId, + [op], + undefined, + undefined, + undefined, + false, + 2, + ); + + expect(staleResults).toEqual([ + expect.objectContaining({ + opId: op.id, + accepted: false, + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }), + ]); + expect(testState.operations.has(op.id)).toBe(false); + expect(testState.userSyncStates.get(userId)?.lastSeq).toBe(4); + expect(testState.userSyncStates.get(userId)?.latestStateReplacementSeq).toBe(3); + + const currentResults = await service.uploadOps( + userId, + clientId, + [op], + undefined, + undefined, + undefined, + false, + 3, + ); + + expect(currentResults).toEqual([ + expect.objectContaining({ + opId: op.id, + accepted: true, + serverSeq: 5, + }), + ]); + expect(testState.operations.has(op.id)).toBe(true); + }); + + it('resolves the latest state replacement for cached upload checks', async () => { + const service = new SyncService(); + const replacement = makeOp({ + id: 'cached-check-state-replacement', + opType: 'BACKUP_IMPORT', + entityType: 'ALL', + entityId: undefined, + }); + testState.operations.set(replacement.id, { + ...replacement, + userId, + serverSeq: 3, + entityId: null, + entityIds: [], + payloadBytes: BigInt(1), + clientTimestamp: BigInt(replacement.timestamp), + receivedAt: BigInt(replacement.timestamp), + isPayloadEncrypted: false, + syncImportReason: null, + repairBaseServerSeq: null, + }); + testState.userSyncStates.set(userId, { + userId, + lastSeq: 4, + latestStateReplacementSeq: null, + }); + + await expect(service.getLatestStateReplacementSeq(userId)).resolves.toBe(3); + await expect(service.getLatestStateReplacementSeq(userId + 1)).resolves.toBeNull(); + }); + + it('persists a resolved no-replacement sentinel on the upload path', async () => { + const service = new SyncService({ batchUpload: true }); + const op = makeOp({ id: 'first-upload-with-cursor' }); + + const result = await service.uploadOps( + userId, + clientId, + [op], + undefined, + undefined, + undefined, + false, + 0, + ); + + expect(result[0].accepted).toBe(true); + expect(testState.userSyncStates.get(userId)?.latestStateReplacementSeq).toBe(0); + }); + it('should correctly upload operations', async () => { const service = getSyncService(); const op: Operation = makeOp(); @@ -1763,6 +1886,7 @@ describe('SyncService', () => { lastSeq: 3, latestFullStateSeq: 2, latestFullStateVectorClock: { [clientId]: 9 }, + latestStateReplacementSeq: 2, }), ); aggregateSpy.mockRestore();