Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-07-29 02:30:03 +00:00
Code Issues Projects Releases Wiki Activity

fix(sync): reject uploads behind state replacements (#9330)

Browse source 
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.
This commit is contained in:
Johannes Millan 2026-07-28 11:28:29 +02:00 • committed by GitHub
parent 6e12545995
commit 2b2a36ae56
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 870 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

5
ARCHITECTURE-DECISIONS.md
View file

@ -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
---

10
docs/sync-and-op-log/operation-log-architecture.md
View file

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

4
docs/wiki/3.01-API.md
View file

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

145
e2e/tests/sync/supersync-encryption-password-change.spec.ts
View file

@ -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<SuperSyncServerOperation[]> => {
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<void> => {
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,

17
packages/super-sync-server/docs/architecture.md
View file

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

5
packages/super-sync-server/prisma/migrations/20260727000000_add_state_replacement_seq/migration.sql Normal file
View file

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

1
packages/super-sync-server/prisma/schema.prisma
View file

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

112
packages/super-sync-server/src/sync/sync.routes.ops-handler.ts
View file

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

132
packages/super-sync-server/src/sync/sync.service.ts
View file

@ -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<string>,
repairBaseServerSeq?: number,
allowLegacyRepairWithoutBase: boolean = false,
lastKnownServerSeq?: number,
): Promise<UploadResult[]> {
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<Array<{ lastSeq: number }>>`
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<number | null> {
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

15
packages/super-sync-server/src/sync/sync.types.ts
View file

@ -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<Pick<Operation, 'id'>>,
): 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,

91
packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts
View file

@ -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 });
},
);
});

11
packages/super-sync-server/tests/migration-sql.spec.ts
View file

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

250
packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts
View file

@ -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 () => {

126
packages/super-sync-server/tests/sync.service.spec.ts
View file

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