Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-07-17 16:37:43 +00:00
Code Issues Projects Releases Wiki Activity

fix(sync): dedupe surgical-sync retries and keep the import author through pruning (#9089)

Browse source 
* fix(sync): deduplicate surgical sync retries

Persist split-file configuration and acknowledge operation IDs already committed remotely after a lost upload response.

* fix(sync): preserve import author during clock pruning

Keep the active causal full-state author in oversized stored clocks and reuse the stored protected IDs when classifying response-loss retries.

* test(sync): harden supersync failure coverage

Exercise real upload endpoints and exact operation IDs across response loss, validation failures, schema blockers, full-state boundaries, vector pruning, and concurrent edits.

* fix(sync): heal a corrupt primary on duplicate-only uploads

The .bak recovery path caches the CORRUPT primary's rev precisely so this
cycle's conditional overwrite repairs sync-ops.json. The duplicate-retry
short-circuit read that cache and returned before the write, leaving the
primary corrupt whenever the recovered buffer already held every pending
op. Flag the recovered entry and let those uploads fall through.

Also stop synthesising a serverSeq for ops already in the buffer: the
field is optional, the upload and download paths number ops differently,
and a mixed batch could hand two ops the same value.

* perf(sync): look the full-state author up once per upload

Batch upload is off by default, so the guarded batch path was not the one
serving production: the serial path queries the causal full-state author
per op inside a single transaction, and a clock of 21-50 entries passes
validation and trips the guard on every one of them. Memoize the author
per transaction and resolve it lazily, so only an op whose clock actually
overflows pays, and only once. This also retires the batch pre-scan and
its loop-carried author, leaving both paths on one mechanism.

Report the lookup through ProcessOperationResult so the upload summary
stops under-reporting round-trips, and record why reconstructing the
stored protected set loosens id-collision detection.

* test(sync): wait for the committed title in renameTask

renameTask blurred the textarea, slept 300ms and returned without ever
checking the rename landed. Blur -> dispatch -> re-render outruns that
delay on a loaded machine, so a following sync uploads without the rename
op and the caller asserts against a task that was never renamed — which
is what supersync 3.1 hits on CI but never locally.

Wait for the new title instead, mirroring markTaskDone's done-state wait
and the e2e no-waitForTimeout rule.

* test(sync): dispatch focus so renameTask actually commits

renameTask relied on el.focus() to emit a focus event, but these tests
drive two clients as separate pages and only one page can hold focus, so
on CI the event often never fires. TaskTitleComponent then keeps
_isFocused=false, and resetToLastExternalValueTrigger resets tmpValue to
the stored title on the next task-object emission. Blur therefore
computes wasChanged=false, task.component skips update(), and the rename
is silently dropped without ever becoming an op.

That is supersync 3.1: client A's rename lives only in tmpValue, A syncs
and uploads nothing, B uploads its done op, A downloads it, the task ref
changes and the title reverts to the original — exactly the state the CI
artifact captured. A real user always has real focus, so the app itself
is unaffected.

Dispatch focus explicitly, mirroring the synthetic input/blur already
used here. Also correct the previous commit's claim: the toBeVisible
wait matches tmpValue, a component-local signal rendered in both
template branches, so it never observed the committed title and could
not have fixed this.

* docs: revert incidental prettier reformat of unrelated docs

The master merge ran prettier across files it pulled in, reformatting
three documents this branch has no business touching: markdown table
cell padding plus *emphasis* -> _emphasis_, with no content change.
handover.md documents two unrelated branches entirely.

Restores them to master. vector-clocks.md keeps its edits — those are
this branch's own and describe the pruning protection.

* refactor(sync): drop the full-state author lookup's roundtrip accounting

resolveFullStateAuthor memoizes per transaction, so the lookup it counts
fires at most once per upload — the plumbing existed to report a number
that is always 0 or 1, on a log line already counting dozens. The memo
and its own accounting cancelled out.

Removing didQuery lets resolveFullStateAuthor return string | undefined
and getPruneProtectedIds return string[], instead of both carrying a
tuple purely to feed the counter.

uploadDbRoundtrips and the batch path's own counter are untouched.

* test(sync): assert the committed store title in renameTask

The focus dispatch did not fix supersync 3.1 — the shard failed again
with an identical snapshot (original title, done, rename gone), so that
diagnosis was wrong. Stop guessing at the trigger and make the helper
able to observe the thing in question.

task-title renders tmpValue, a component-local signal, in BOTH its
editing and idle branches. Every DOM assertion here therefore matches as
soon as the synthetic input event fires, whether or not an op was ever
captured — which is why two rounds of "wait for the title" changed
nothing. Read the store instead, via the __e2eTestHelpers.store hook the
timeSpent helper already uses.

This is a diagnostic as much as a fix: it splits the two remaining
explanations. If renameTask now fails, the rename never becomes an op
and the bug is in how the test drives the edit. If it passes and 3.1
still fails at the merge assertion, the op is captured and lost during
sync — a real defect, and the test is right to fail.

* test(sync): move the 3.1 disjoint-merge rewrite out to #9095

3.1 was the last red shard, and it turned out to be right: the
store-backed renameTask passes, so the rename IS captured as an op, and
the test still fails at the merge assertion — the op is committed and
then lost during sync. Filed as #9095.

That bug is pre-existing and cannot be reached by anything in this PR:
the file-based adapter is not used by SuperSync, and the server-side
author memo only engages for clocks over 20 entries where this test
carries about three. 3.1 is also the only test here that exercises
neither of this PR's fixes — it races a title change against a
move-to-done, which is conflict resolution, not retry dedup or clock
pruning. So it moves to #9095 rather than holding verified sync fixes
red.

The rest of the hardening stays: the fault injections whose globs never
matched a real endpoint, the schema-mismatch test that asserted nothing,
and the compaction suite that called an endpoint which never existed are
what actually cover the fixes here.

Restoring the old 3.1 puts a misleading test back, so it now carries a
comment saying why it proves little and where the real one lives. The
strengthened version is kept on test/issue-9095-disjoint-merge-repro.

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
This commit is contained in:
Johannes Millan 2026-07-17 00:46:01 +02:00 • committed by GitHub
parent 0238b5a729
commit b50d5f6d96
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1377 additions and 599 deletions
Download patch file Download diff file Expand all files Collapse all files

45
docs/sync-and-op-log/vector-clocks.md
View file

@ -108,7 +108,7 @@ In `sync.service.ts` (`processOperation`):
1. `ValidationService.validateOp()` sanitizes the clock (DoS cap at 2.5×MAX = 50 entries) but does **NOT** prune
2. `detectConflict()` compares the **full unpruned** incoming clock against the existing entity clock
3. If accepted: `limitVectorClockSize(clock, [clientId])` prunes to MAX before storage, preserving only the uploading client's ID
3. If accepted: `limitVectorClockSize()` prunes to MAX before storage, preserving the uploading client and, when present, the latest causal full-state author
4. The pruned clock is stored in the database
### Step 3: Download by Other Clients
@ -146,22 +146,22 @@ Implemented in `packages/sync-core/src/vector-clock.ts`. The client wrapper in `
### When Pruning Happens (Exhaustive List)
| Location | When | What's Preserved |
| ----------------------------------------------- | ----------------------------------------------- | --------------------- |
| **Server** `processOperation()` | After conflict detection, before storage | Uploading client's ID |
| **Server** `getOpsSinceWithSeq()` | Aggregating snapshot vector clock | Requesting client |
| **Client** `SyncHydrationService` | Creating SYNC_IMPORT during conflict resolution | Current client only |
| **Client** `ServerMigrationService` | Creating SYNC_IMPORT during migration | Current client only |
| **Client** `RepairOperationService` | Creating REPAIR operation | Current client only |
| **Client** `OperationLogSnapshotService` | Saving snapshot to state cache | Current client only |
| **Client** `OperationLogCompactionService` | Compaction (saving snapshot + deleting old ops) | Current client only |
| **Client** `OperationLogHydratorService` | Restoring snapshot during hydration | Current client only |
| **Client** normal op capture | **NEVER** | N/A |
| **Client** `SupersededOperationResolverService` | **NEVER** (conflict resolution) | N/A |
| Location | When | What's Preserved |
| ----------------------------------------------- | ----------------------------------------------- | ------------------------------------------- |
| **Server** `processOperation()` | After conflict detection, before storage | Uploading client + active full-state author |
| **Server** `getOpsSinceWithSeq()` | Aggregating snapshot vector clock | Requesting client |
| **Client** `SyncHydrationService` | Creating SYNC_IMPORT during conflict resolution | Current client only |
| **Client** `ServerMigrationService` | Creating SYNC_IMPORT during migration | Current client only |
| **Client** `RepairOperationService` | Creating REPAIR operation | Current client only |
| **Client** `OperationLogSnapshotService` | Saving snapshot to state cache | Current client only |
| **Client** `OperationLogCompactionService` | Compaction (saving snapshot + deleting old ops) | Current client only |
| **Client** `OperationLogHydratorService` | Restoring snapshot during hydration | Current client only |
| **Client** normal op capture | **NEVER** | N/A |
| **Client** `SupersededOperationResolverService` | **NEVER** (conflict resolution) | N/A |
### Pruning is Rare
With MAX=20, a user needs 21+ unique client IDs before pruning triggers. In the unlikely event it does trigger, the worst case is one extra server round-trip (false CONCURRENT → client resolves → re-uploads with >MAX clock → GREATER_THAN → accepted).
With MAX=20, a user needs 21+ unique client IDs before pruning triggers. The server preserves both the uploader and the latest causal full-state author when they are present in the incoming clock. Preserving that boundary edge prevents high-counter historical clients from making a post-import operation appear concurrent to the importing client. Other pruned edges can still cause one extra server round-trip (false CONCURRENT → client resolves → re-uploads with >MAX clock → GREATER_THAN → accepted).
---
@ -406,16 +406,21 @@ before storage**. This is the invariant in §6 and §9.
### Why MAX = 20 (the 10 → 30 → 20 evolution)
The original defense against the Feb-2026 loop was a 4-layer scheme (protected
client IDs, pruning-aware comparison, an `isLikelyPruningArtifact` heuristic,
the same-client check) — symptom treatment. The root cause was that `MAX = 10`
was too small, making pruning frequent and interacting badly with SYNC_IMPORT.
The original defense against the Feb-2026 loop was a 4-layer scheme (broad
protected-client tracking, pruning-aware comparison, an
`isLikelyPruningArtifact` heuristic, the same-client check) — symptom treatment.
The root cause was that `MAX = 10` was too small, making pruning frequent and
interacting badly with SYNC_IMPORT.
Commit `d70f18a94d` raised `MAX` 10 → 30 (later reduced to 20 — a 20-entry
clock is ~333 bytes, negligible) and **removed three of the four layers**.
clock is ~333 bytes, negligible) and removed the broad tracking and comparison
heuristics. The server now has one narrow storage-only exception: it preserves
the latest causal full-state author so post-import operations retain that
boundary edge.
`isLikelyPruningArtifact` was dropped (known false positives, unnecessary at
MAX = 20). Only the **same-client check** remains — always mathematically
correct (monotonic counters are definitive) and independent of MAX. At
correct during conflict comparison (monotonic counters are definitive) and
independent of MAX. At
MAX = 20, pruning needs **21+ distinct client IDs**, extremely rare for a
personal productivity app, so the pruning path is effectively dormant
(see §5 "Pruning is Rare").

12
e2e/pages/sync.page.ts
View file

@ -47,6 +47,7 @@ export class SyncPage extends BasePage {
syncFolderPath: string;
isEncryptionEnabled?: boolean;
encryptionPassword?: string;
isUseSplitSyncFiles?: boolean;
/**
* Set the encryption password in the setup-time "Encrypt before first
* upload?" dialog (instead of the post-setup Enable Encryption button), so
@ -211,6 +212,17 @@ export class SyncPage extends BasePage {
await this.passwordInput.fill(config.password);
await this.syncFolderInput.fill(config.syncFolderPath);
if (config.isUseSplitSyncFiles !== undefined) {
await this.expandAdvancedSettings();
const splitSyncCheckbox = dialog.getByRole('checkbox', {
name: /Surgical sync/i,
});
await splitSyncCheckbox.setChecked(config.isUseSplitSyncFiles);
await expect(splitSyncCheckbox).toBeChecked({
checked: config.isUseSplitSyncFiles,
});
}
// Save the configuration
await this.saveBtn.click();

166
e2e/tests/sync/supersync-compaction.spec.ts
View file

@ -1,4 +1,5 @@
import { test } from '../../fixtures/supersync.fixture';
import { test, expect } from '../../fixtures/supersync.fixture';
import type { Page } from '@playwright/test';
import {
createTestUser,
getSuperSyncConfig,
@ -12,10 +13,10 @@ import {
} from '../../utils/supersync-helpers';
/**
* SuperSync Compaction/Snapshot Resilience E2E Tests
* SuperSync Full-State Boundary Resilience E2E Tests
*
* Tests that server snapshot creation doesn't break sync for
* existing or new clients.
* Tests that a real full-state replacement doesn't break sync for existing or
* new clients.
*
* Prerequisites:
* - super-sync-server running on localhost:1901 with TEST_MODE=true
@ -24,52 +25,81 @@ import {
* Run with: npm run e2e:supersync:file e2e/tests/sync/supersync-compaction.spec.ts
*/
/**
* Trigger server-side snapshot generation via the test API.
* Falls back gracefully if endpoint doesn't exist.
*/
const triggerServerSnapshot = async (token: string): Promise<boolean> => {
try {
const response = await fetch(`${SUPERSYNC_BASE_URL}/api/test/trigger-snapshot`, {
method: 'POST',
headers: {
// eslint-disable-next-line @typescript-eslint/naming-convention
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
});
if (response.ok) {
console.log('[Compaction] Server snapshot triggered successfully');
return true;
}
console.log(
`[Compaction] Snapshot trigger returned ${response.status} - endpoint may not exist`,
interface ServerOperationSummary {
id: string;
opType: string;
serverSeq: number;
}
const getServerOperations = async (userId: number): Promise<ServerOperationSummary[]> => {
const response = await fetch(
`${SUPERSYNC_BASE_URL}/api/test/user/${userId}/ops?limit=100`,
);
if (!response.ok) {
throw new Error(
`Failed to inspect server operations: ${response.status} ${await response.text()}`,
);
return false;
} catch {
console.log('[Compaction] Snapshot trigger failed - endpoint may not exist');
return false;
}
const body = (await response.json()) as { ops: ServerOperationSummary[] };
return body.ops;
};
test.describe('@supersync Compaction/Snapshot Resilience', () => {
const getTaskDoneStates = async (
page: Page,
taskNames: string[],
): Promise<Record<string, boolean>> =>
page.evaluate((names) => {
type TaskLike = { title?: string; isDone?: boolean };
type StoreState = {
task?: { entities?: Record<string, TaskLike | undefined> };
tasks?: { entities?: Record<string, TaskLike | undefined> };
};
type StoreLike = {
subscribe: (next: (state: StoreState) => void) => { unsubscribe: () => void };
};
const store = (window as unknown as { __e2eTestHelpers?: { store?: StoreLike } })
.__e2eTestHelpers?.store;
if (!store) {
throw new Error('__e2eTestHelpers.store missing');
}
let latestState: StoreState | undefined;
const subscription = store.subscribe((state) => {
latestState = state;
});
subscription.unsubscribe();
const taskEntities = [
...Object.values(latestState?.tasks?.entities ?? {}),
...Object.values(latestState?.task?.entities ?? {}),
];
return names.reduce<Record<string, boolean>>((states, name) => {
const task = taskEntities.find((candidate) => candidate?.title?.includes(name));
if (task) {
states[name] = task.isDone === true;
}
return states;
}, {});
}, taskNames);
test.describe('@supersync Full-State Boundary Resilience', () => {
/**
* Scenario: Heavy operations, snapshot, then new client joins
* Scenario: Heavy operations, full-state replacement, then new client joins
*
* This tests that after many operations and a potential snapshot,
* a new client can still get complete state and bidirectional sync
* continues working.
* This tests that after many operations and a full-state replacement, a new
* client can still get complete state and bidirectional sync continues working.
*
* Flow:
* 1. Client A creates 10+ tasks, marks some done, renames some
* 2. Client A syncs
* 3. Attempt to trigger server snapshot (may not be available)
* 4. Client A creates more tasks post-snapshot
* 3. Rotate encryption, which replaces the operation history with a real
* encrypted SYNC_IMPORT full-state boundary
* 4. Client A creates more tasks after the boundary
* 5. New Client B joins and syncs
* 6. Verify B gets complete state (pre + post snapshot data)
* 6. Verify B gets complete state (pre + post-boundary data)
* 7. Bidirectional sync continues working
*/
test('New client receives complete state after heavy operations', async ({
test('New client receives complete state after full-state replacement', async ({
browser,
baseURL,
testRunId,
@ -114,21 +144,27 @@ test.describe('@supersync Compaction/Snapshot Resilience', () => {
await clientA.sync.syncAndWait();
console.log('[Compaction] Client A synced all operations');
// ============ PHASE 2: Attempt to trigger snapshot ============
console.log('[Compaction] Phase 2: Attempting server snapshot');
// ============ PHASE 2: Create a real full-state boundary ============
console.log('[Compaction] Phase 2: Creating full-state boundary');
const snapshotTriggered = await triggerServerSnapshot(user.token);
if (!snapshotTriggered) {
console.log(
'[Compaction] Snapshot endpoint not available - testing without snapshot',
);
console.log(
'[Compaction] (Test still validates heavy-operation sync resilience)',
);
}
const operationsBeforeBoundary = await getServerOperations(user.userId);
expect(operationsBeforeBoundary.length).toBeGreaterThan(1);
const maxSeqBeforeBoundary = Math.max(
...operationsBeforeBoundary.map((op) => op.serverSeq),
);
// ============ PHASE 3: Client A creates more tasks after snapshot ============
console.log('[Compaction] Phase 3: Creating post-snapshot tasks');
const postBoundaryPassword = `post-boundary-${testRunId}`;
await clientA.sync.changeEncryptionPassword(postBoundaryPassword);
const operationsAtBoundary = await getServerOperations(user.userId);
expect(operationsAtBoundary).toHaveLength(1);
expect(operationsAtBoundary[0].opType).toBe('SYNC_IMPORT');
expect(operationsAtBoundary[0].serverSeq).toBeGreaterThan(maxSeqBeforeBoundary);
const boundarySeq = operationsAtBoundary[0].serverSeq;
console.log(`[Compaction] Full-state boundary stored at serverSeq ${boundarySeq}`);
// ============ PHASE 3: Client A creates more tasks after boundary ============
console.log('[Compaction] Phase 3: Creating post-boundary tasks');
const postSnapshotTask1 = `PostSnap-1-${testRunId}`;
const postSnapshotTask2 = `PostSnap-2-${testRunId}`;
@ -136,13 +172,29 @@ test.describe('@supersync Compaction/Snapshot Resilience', () => {
await clientA.workView.addTask(postSnapshotTask2);
await clientA.sync.syncAndWait();
console.log('[Compaction] Client A synced post-snapshot tasks');
console.log('[Compaction] Client A synced post-boundary tasks');
const operationsAfterBoundary = await getServerOperations(user.userId);
expect(
operationsAfterBoundary.some(
(op) => op.opType === 'SYNC_IMPORT' && op.serverSeq === boundarySeq,
),
).toBe(true);
expect(
operationsAfterBoundary.filter((op) => op.serverSeq > boundarySeq).length,
).toBeGreaterThanOrEqual(2);
expect(operationsAfterBoundary.every((op) => op.serverSeq >= boundarySeq)).toBe(
true,
);
// ============ PHASE 4: New Client B joins ============
console.log('[Compaction] Phase 4: New Client B joins');
clientB = await createSimulatedClient(browser, appUrl, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
await clientB.sync.setupSuperSync({
...syncConfig,
password: postBoundaryPassword,
});
// Client B syncs to get all data
await clientB.sync.syncAndWait();
@ -153,12 +205,18 @@ test.describe('@supersync Compaction/Snapshot Resilience', () => {
// ============ PHASE 5: Verify B gets complete state ============
console.log('[Compaction] Phase 5: Verifying Client B has complete state');
const taskDoneStates = await getTaskDoneStates(clientB.page, taskNames);
expect(Object.keys(taskDoneStates)).toHaveLength(taskNames.length);
for (let i = 0; i < taskNames.length; i++) {
expect(taskDoneStates[taskNames[i]]).toBe(i < 3);
}
// Check undone tasks are visible (first 3 were marked done)
for (let i = 3; i < taskNames.length; i++) {
await waitForTask(clientB.page, taskNames[i]);
}
// Check post-snapshot tasks
// Check post-boundary tasks
await waitForTask(clientB.page, postSnapshotTask1);
await waitForTask(clientB.page, postSnapshotTask2);
console.log('[Compaction] ✓ Client B received all expected tasks');
@ -177,7 +235,7 @@ test.describe('@supersync Compaction/Snapshot Resilience', () => {
await waitForTask(clientA.page, taskFromB);
console.log("[Compaction] ✓ Client A received Client B's task");
console.log('[Compaction] ✓ Compaction/snapshot resilience test PASSED!');
console.log('[Compaction] ✓ Full-state boundary resilience test PASSED!');
} finally {
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);

425
e2e/tests/sync/supersync-error-scenarios.spec.ts
View file

@ -1,13 +1,77 @@
import { test, expect } from '../../fixtures/supersync.fixture';
import type { Page } from '@playwright/test';
import {
createTestUser,
getSuperSyncConfig,
createSimulatedClient,
closeClient,
parseSuperSyncRequestBody,
routeSuperSyncOps,
unrouteSuperSyncOps,
waitForTask,
type SimulatedE2EClient,
} from '../../utils/supersync-helpers';
interface MutableServerOperation {
op?: {
id?: string;
schemaVersion?: number;
};
}
interface OperationDownloadBody {
ops?: MutableServerOperation[];
}
interface OperationUploadBody {
ops: Array<{ id: string }>;
}
const getSuperSyncCursor = async (page: Page): Promise<string | null> =>
page.evaluate(() => {
const key = Object.keys(localStorage).find((candidate) =>
candidate.startsWith('super_sync_last_server_seq_'),
);
return key ? localStorage.getItem(key) : null;
});
const areLocalOperationsSynced = (page: Page, operationIds: string[]): Promise<boolean> =>
page.evaluate(async (ids) => {
const db = await new Promise<IDBDatabase>((resolve, reject) => {
const openRequest = indexedDB.open('SUP_OPS');
openRequest.onsuccess = (): void => resolve(openRequest.result);
openRequest.onerror = (): void => reject(openRequest.error);
});
try {
const entries = await Promise.all(
ids.map(
(id) =>
new Promise<{ syncedAt?: number; rejectedAt?: number } | undefined>(
(resolve, reject) => {
const tx = db.transaction('ops', 'readonly');
const getRequest = tx.objectStore('ops').index('byId').get(id);
getRequest.onsuccess = (): void =>
resolve(
getRequest.result as
| { syncedAt?: number; rejectedAt?: number }
| undefined,
);
getRequest.onerror = (): void => reject(getRequest.error);
},
),
),
);
return (
ids.length > 0 &&
entries.every(
(entry) => entry?.syncedAt !== undefined && entry.rejectedAt === undefined,
)
);
} finally {
db.close();
}
}, operationIds);
/**
* SuperSync Error Scenarios E2E Tests
*
@ -39,6 +103,8 @@ test.describe('@supersync Error Scenarios', () => {
test.setTimeout(90000);
let clientA: SimulatedE2EClient | null = null;
const state = { interceptUpload: true };
const rejectedOpIds: string[] = [];
const subsequentUploadIds: string[] = [];
try {
const user = await createTestUser(testRunId);
@ -52,36 +118,34 @@ test.describe('@supersync Error Scenarios', () => {
await clientA.workView.addTask(taskName);
await waitForTask(clientA.page, taskName);
// Intercept the upload to return a VALIDATION_ERROR rejection
// NOTE: With mandatory encryption, POST body is encrypted binary, not JSON.
// We forward the request to get real op IDs from the server response,
// then return the rejection.
await clientA.page.route('**/api/sync/ops', async (route) => {
// Intercept the upload and reject the exact operation IDs emitted by the client.
await routeSuperSyncOps(clientA.page, async (route) => {
if (state.interceptUpload && route.request().method() === 'POST') {
state.interceptUpload = false;
console.log('[Test] Simulating VALIDATION_ERROR rejection');
// Forward request to server to get real response with op IDs
const response = await route.fetch();
const realBody = await response.json().catch(() => ({}));
// Use a fake op ID since we can't parse encrypted request body
const results = [
{
opId: realBody?.results?.[0]?.opId || 'fake-op-id',
accepted: false,
error: 'Invalid entity structure',
errorCode: 'VALIDATION_ERROR',
},
];
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
rejectedOpIds.push(...upload.ops.map((operation) => operation.id));
expect(rejectedOpIds.length).toBeGreaterThan(0);
const results = upload.ops.map((operation) => ({
opId: operation.id,
accepted: false,
error: 'Invalid entity structure',
errorCode: 'VALIDATION_ERROR',
}));
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({
results,
latestSeq: realBody?.latestSeq || 1,
latestSeq: 0,
}),
});
} else if (route.request().method() === 'POST') {
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
subsequentUploadIds.push(...upload.ops.map((operation) => operation.id));
await route.continue();
} else {
await route.continue();
}
@ -95,9 +159,6 @@ test.describe('@supersync Error Scenarios', () => {
// Expected — triggerSync may throw on error state
}
// Remove interception
await clientA.page.unroute('**/api/sync/ops');
// Verify sync shows error status (permanentRejectionCount > 0 → ERROR)
const hasError = await clientA.sync.hasSyncError();
expect(hasError).toBe(true);
@ -105,13 +166,15 @@ test.describe('@supersync Error Scenarios', () => {
// Sync again — the rejected op should NOT be retried
// (it should sync successfully since the rejected op is skipped)
await clientA.sync.syncAndWait();
expect(subsequentUploadIds).not.toEqual(expect.arrayContaining(rejectedOpIds));
await unrouteSuperSyncOps(clientA.page);
console.log(
'[ValidationError] Validation error correctly caused error status and op was not retried',
);
} finally {
if (clientA) {
await clientA.page.unroute('**/api/sync/ops').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
await closeClient(clientA);
}
}
@ -129,6 +192,7 @@ test.describe('@supersync Error Scenarios', () => {
}) => {
test.setTimeout(60000);
let clientA: SimulatedE2EClient | null = null;
const rejectedOpIds: string[] = [];
try {
const user = await createTestUser(testRunId);
@ -155,36 +219,23 @@ test.describe('@supersync Error Scenarios', () => {
// Intercept upload to return rejected ops with "Payload too large" error.
// The app shows alertDialog only when rejected ops contain this text,
// not on raw HTTP 413 responses.
// We forward the request to get real op IDs from the server response.
await clientA.page.route('**/api/sync/ops', async (route) => {
await routeSuperSyncOps(clientA.page, async (route) => {
if (route.request().method() === 'POST') {
console.log('[Test] Simulating Payload Too Large rejection');
const response = await route.fetch();
const realBody = await response.json().catch(() => ({}));
// Use real op IDs so the app can look up the ops in its local store
const realResults = (realBody?.results || []) as Array<{
opId: string;
accepted: boolean;
}>;
const rejectedResults = realResults.map((r) => ({
opId: r.opId,
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
rejectedOpIds.push(...upload.ops.map((operation) => operation.id));
expect(rejectedOpIds.length).toBeGreaterThan(0);
const rejectedResults = upload.ops.map((operation) => ({
opId: operation.id,
accepted: false,
error: 'Payload too large',
}));
// Fallback if no results from server
if (rejectedResults.length === 0) {
rejectedResults.push({
opId: 'fake-op-id',
accepted: false,
error: 'Payload too large',
});
}
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({
results: rejectedResults,
latestSeq: realBody?.latestSeq || 1,
latestSeq: 0,
}),
});
} else {
@ -212,6 +263,7 @@ test.describe('@supersync Error Scenarios', () => {
}
// Verify alert was shown with appropriate message
expect(rejectedOpIds.length).toBeGreaterThan(0);
expect(alertShown).toBe(true);
expect(alertMessage.length).toBeGreaterThan(0);
@ -221,7 +273,7 @@ test.describe('@supersync Error Scenarios', () => {
console.log('[PayloadTooLarge] Alert dialog shown for 413 response');
} finally {
if (clientA) {
await clientA.page.unroute('**/api/sync/ops').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
await closeClient(clientA);
}
}
@ -243,6 +295,7 @@ test.describe('@supersync Error Scenarios', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { returnDuplicate: false };
const duplicateOpIds: string[] = [];
try {
const user = await createTestUser(testRunId);
@ -262,31 +315,31 @@ test.describe('@supersync Error Scenarios', () => {
await waitForTask(clientA.page, taskName2);
// Intercept the next upload to return DUPLICATE_OPERATION
// NOTE: With mandatory encryption, POST body is encrypted binary, not JSON.
// We forward the request to get real response, then return the rejection.
// The server accepts the upload, but the client receives a duplicate response
// for the exact IDs it sent (the lost-acknowledgement recovery case).
state.returnDuplicate = true;
await clientA.page.route('**/api/sync/ops', async (route) => {
await routeSuperSyncOps(clientA.page, async (route) => {
if (state.returnDuplicate && route.request().method() === 'POST') {
state.returnDuplicate = false;
console.log('[Test] Simulating DUPLICATE_OPERATION rejection');
// Forward request to server to get real response
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
duplicateOpIds.push(...upload.ops.map((operation) => operation.id));
expect(duplicateOpIds.length).toBeGreaterThan(0);
const response = await route.fetch();
const realBody = await response.json().catch(() => ({}));
const realBody = (await response.json()) as { latestSeq?: number };
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({
results: [
{
opId: realBody?.results?.[0]?.opId || 'fake-op-id',
accepted: false,
error: 'Duplicate operation',
errorCode: 'DUPLICATE_OPERATION',
},
],
latestSeq: realBody?.latestSeq || 2,
results: upload.ops.map((operation) => ({
opId: operation.id,
accepted: false,
error: 'Duplicate operation',
errorCode: 'DUPLICATE_OPERATION',
})),
latestSeq: realBody.latestSeq ?? 0,
}),
});
} else {
@ -302,8 +355,14 @@ test.describe('@supersync Error Scenarios', () => {
// May or may not throw
}
await expect
.poll(() => areLocalOperationsSynced(clientA!.page, duplicateOpIds), {
message: 'the duplicate response must acknowledge the exact uploaded ops',
})
.toBe(true);
// Remove interception
await clientA.page.unroute('**/api/sync/ops');
await unrouteSuperSyncOps(clientA.page);
// Verify no error shown — duplicate should be handled silently
// After removing the route, the next sync should succeed
@ -319,11 +378,12 @@ test.describe('@supersync Error Scenarios', () => {
await clientB.sync.setupSuperSync(syncConfig);
await clientB.sync.syncAndWait();
await waitForTask(clientB.page, taskName);
await waitForTask(clientB.page, taskName2);
console.log('[DuplicateOp] Duplicate operation handled silently without error');
} finally {
if (clientA) {
await clientA.page.unroute('**/api/sync/ops').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
await closeClient(clientA);
}
if (clientB) await closeClient(clientB);
@ -331,12 +391,9 @@ test.describe('@supersync Error Scenarios', () => {
});
/**
* Scenario G.7: Schema version mismatch returns handled error
*
* When downloaded ops have a modelVersion higher than the client's,
* the client should log a warning and return HANDLED_ERROR without crashing.
* Scenario G.7: A future-schema operation blocks cursor advancement
*/
test('Schema version mismatch returns handled error without crash', async ({
test('Future-schema operation blocks until a compatible response is available', async ({
browser,
baseURL,
testRunId,
@ -344,169 +401,181 @@ test.describe('@supersync Error Scenarios', () => {
test.setTimeout(90000);
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { injectFutureSchemaOps: true };
let injectedResponses = 0;
try {
const user = await createTestUser(testRunId);
const syncConfig = getSuperSyncConfig(user);
// Client A creates real data
// Establish a shared cursor before creating the incompatible suffix.
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const taskName = `Schema-${testRunId}`;
await clientA.workView.addTask(taskName);
await clientA.sync.syncAndWait();
// Client B will receive ops with future schema version
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
const cursorBeforeBlock = await getSuperSyncCursor(clientB.page);
expect(cursorBeforeBlock).not.toBeNull();
// Intercept download to inject ops with a very high schema version
await clientB.page.route('**/api/sync/ops/**', async (route) => {
if (route.request().method() === 'GET' && state.injectFutureSchemaOps) {
state.injectFutureSchemaOps = false;
console.log('[Test] Injecting ops with future schema version');
// Get real response and modify it
const response = await route.fetch();
const json = await response.json();
// Modify all ops to have a very high schema version
if (json.ops) {
for (const op of json.ops) {
op.schemaVersion = 99999;
}
}
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(json),
});
} else {
await route.continue();
const taskName = `Schema-${testRunId}`;
const uploadedOpIds: string[] = [];
await routeSuperSyncOps(clientA.page, async (route) => {
if (route.request().method() === 'POST') {
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
uploadedOpIds.push(...upload.ops.map((operation) => operation.id));
}
await route.continue();
});
await clientA.workView.addTask(taskName);
await clientA.sync.syncAndWait();
await unrouteSuperSyncOps(clientA.page);
expect(uploadedOpIds).toHaveLength(1);
const blockerOpId = uploadedOpIds[0];
// Tamper with the real operation wrapper on every retry. `schemaVersion`
// is plaintext metadata beside the encrypted payload.
await routeSuperSyncOps(clientB.page, async (route) => {
if (route.request().method() === 'GET') {
const response = await route.fetch();
const body = (await response.json()) as OperationDownloadBody;
const blocker = body.ops?.find(({ op }) => op?.id === blockerOpId);
if (blocker?.op) {
// 99 is within the transport contract (1..100) but newer than this
// client's current schema, so it exercises the sync-layer blocker.
blocker.op.schemaVersion = 99;
injectedResponses++;
}
await route.fulfill({
response,
body: JSON.stringify(body),
});
return;
}
await route.continue();
});
// Client B syncs — should handle the schema mismatch gracefully
try {
await clientB.sync.triggerSync();
await clientB.page.waitForTimeout(3000);
} catch {
// May or may not throw depending on error handling
}
// `triggerSync()` is a success-oriented helper and may throw as soon as
// the expected error icon appears. Click directly and require the stable
// blocked state before inspecting the cursor.
await clientB.sync.syncBtn.click();
await expect.poll(() => clientB!.sync.hasSyncError()).toBe(true);
// Remove interception and retry with real data
await clientB.page.unroute('**/api/sync/ops/**');
expect(injectedResponses).toBeGreaterThan(0);
expect(await getSuperSyncCursor(clientB.page)).toBe(cursorBeforeBlock);
await expect(
clientB.page.locator(`task:has-text("${taskName}")`),
).not.toBeVisible();
await unrouteSuperSyncOps(clientB.page);
await clientB.sync.syncAndWait();
// Verify Client B didn't crash and can still sync
await waitForTask(clientB.page, taskName);
const hasError = await clientB.sync.hasSyncError();
expect(hasError).toBe(false);
console.log(
'[SchemaVersionMismatch] Client handled future schema version without crash',
);
expect(await clientB.sync.hasSyncError()).toBe(false);
expect(await getSuperSyncCursor(clientB.page)).not.toBe(cursorBeforeBlock);
} finally {
if (clientA) await closeClient(clientA);
if (clientA) {
await unrouteSuperSyncOps(clientA.page).catch(() => {});
await closeClient(clientA);
}
if (clientB) {
await clientB.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientB.page).catch(() => {});
await closeClient(clientB);
}
}
});
/**
* Scenario G.8: Failed operation migration skips op and other ops still apply
*
* When a downloaded op has a corrupted/unmigrateable structure,
* it should be skipped and other valid ops should still be applied.
* Scenario G.8: A mid-batch schema blocker applies only the valid prefix and keeps
* the cursor before the blocker so the suffix can be retried after recovery.
*/
test('Failed operation migration skips corrupted op, applies others', async ({
test('Mid-batch schema blocker applies prefix and retries suffix from prior cursor', async ({
browser,
baseURL,
testRunId,
}) => {
test.setTimeout(90000);
test.setTimeout(120000);
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { injectCorruptedOp: true };
let injectedResponses = 0;
try {
const user = await createTestUser(testRunId);
const syncConfig = getSuperSyncConfig(user);
// Client A creates real data
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const taskName = `MigrationFail-${testRunId}`;
await clientA.workView.addTask(taskName);
await clientA.sync.syncAndWait();
// Client B will receive ops including one corrupted one
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
const cursorBeforeBlock = await getSuperSyncCursor(clientB.page);
expect(cursorBeforeBlock).not.toBeNull();
// Intercept download to inject a corrupted op alongside valid ones
await clientB.page.route('**/api/sync/ops/**', async (route) => {
if (route.request().method() === 'GET' && state.injectCorruptedOp) {
state.injectCorruptedOp = false;
console.log('[Test] Injecting corrupted op into download response');
const response = await route.fetch();
const json = await response.json();
// Insert a corrupted op before the valid ones
if (json.ops && json.ops.length > 0) {
const corruptedOp = {
id: 'corrupted-migration-op',
opType: 'UPD',
entityType: 'TASK',
entityId: 'nonexistent-entity',
actionType: '[Task] CORRUPTED_ACTION',
payload: { title: undefined, __broken: true },
vectorClock: { broken_client: 1 },
timestamp: Date.now(),
schemaVersion: 0, // Very old schema, likely to fail migration
clientId: 'broken-client',
};
json.ops.unshift(corruptedOp);
}
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(json),
});
} else {
await route.continue();
const taskNames = [
`MigrationPrefix-${testRunId}`,
`MigrationBlocker-${testRunId}`,
`MigrationSuffix-${testRunId}`,
];
const uploadedOpIds: string[] = [];
await routeSuperSyncOps(clientA.page, async (route) => {
if (route.request().method() === 'POST') {
const upload = parseSuperSyncRequestBody<OperationUploadBody>(route.request());
uploadedOpIds.push(...upload.ops.map((operation) => operation.id));
}
await route.continue();
});
for (const taskName of taskNames) {
await clientA.workView.addTask(taskName);
}
await clientA.sync.syncAndWait();
await unrouteSuperSyncOps(clientA.page);
expect(uploadedOpIds).toHaveLength(3);
const blockerOpId = uploadedOpIds[1];
await routeSuperSyncOps(clientB.page, async (route) => {
if (route.request().method() === 'GET') {
const response = await route.fetch();
const body = (await response.json()) as OperationDownloadBody;
const blocker = body.ops?.find(({ op }) => op?.id === blockerOpId);
if (blocker?.op) {
// Use a real encrypted operation and change only its schema metadata,
// preserving real server sequences on both sides of the blocker.
blocker.op.schemaVersion = 99;
injectedResponses++;
}
await route.fulfill({
response,
body: JSON.stringify(body),
});
return;
}
await route.continue();
});
// Client B syncs — corrupted op should be skipped, valid ops applied
await clientB.sync.syncBtn.click();
await expect.poll(() => clientB!.sync.hasSyncError()).toBe(true);
expect(injectedResponses).toBeGreaterThan(0);
expect(await getSuperSyncCursor(clientB.page)).toBe(cursorBeforeBlock);
await waitForTask(clientB.page, taskNames[0]);
await expect(
clientB.page.locator(`task:has-text("${taskNames[1]}")`),
).not.toBeVisible();
await expect(
clientB.page.locator(`task:has-text("${taskNames[2]}")`),
).not.toBeVisible();
await unrouteSuperSyncOps(clientB.page);
await clientB.sync.syncAndWait();
// Remove interception
await clientB.page.unroute('**/api/sync/ops/**');
// Verify the valid task was still received despite the corrupted op
await waitForTask(clientB.page, taskName);
// Verify Client B is healthy and can sync again
await clientB.sync.syncAndWait();
const hasError = await clientB.sync.hasSyncError();
expect(hasError).toBe(false);
console.log(
'[MigrationFailure] Corrupted op skipped, valid ops applied successfully',
);
for (const taskName of taskNames) {
await waitForTask(clientB.page, taskName);
}
expect(await clientB.sync.hasSyncError()).toBe(false);
expect(await getSuperSyncCursor(clientB.page)).not.toBe(cursorBeforeBlock);
} finally {
if (clientA) await closeClient(clientA);
if (clientA) {
await unrouteSuperSyncOps(clientA.page).catch(() => {});
await closeClient(clientA);
}
if (clientB) {
await clientB.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientB.page).catch(() => {});
await closeClient(clientB);
}
}

225
e2e/tests/sync/supersync-import-other-client-ops.spec.ts
View file

@ -4,6 +4,10 @@ import {
getSuperSyncConfig,
createSimulatedClient,
closeClient,
parseSuperSyncRequestBody,
routeSuperSyncOps,
SUPERSYNC_BASE_URL,
unrouteSuperSyncOps,
waitForTask,
type SimulatedE2EClient,
} from '../../utils/supersync-helpers';
@ -14,47 +18,70 @@ import { expectTaskOnAllClients } from '../../utils/supersync-assertions';
import { waitForAppReady } from '../../utils/waits';
/**
* SuperSync: Other Client's Post-Import Operations (Vector Clock Bloat Bug)
* SuperSync: Other Client's Post-Import Operations Across Clock Pruning
*
* These tests verify that operations created by a DIFFERENT client than the one
* that performed a SYNC_IMPORT are correctly synced back to the importing client.
*
* BUG SCENARIO (vector clock merge-vs-replace):
* PRUNING SCENARIO:
* 1. Client B has accumulated a large vector clock from long history (many old devices)
* 2. Client A does SYNC_IMPORT with a fresh clock
* 3. Client B receives the SYNC_IMPORT — mergeRemoteOpClocks() MERGES the import's
* fresh clock into B's old accumulated clock instead of REPLACING it
* 4. B's clock now has 13+ entries (old entries + import's entry)
* 5. B creates new ops — their clocks carry all 13+ entries
* 6. Server prunes B's ops to MAX=20, dropping import's entry (lowest counter)
* 7. Client A receives B's pruned ops — compareVectorClocks returns CONCURRENT
* (B has old entries A doesn't know, A has import entry B lost to pruning)
* 8. SyncImportFilterService filters B's ops as "concurrent with import"
* 9. Client A never sees Client B's new tasks
* 4. B's clock now has 21+ entries (old entries + import's entry)
* 5. B creates new ops — their clocks carry all 21+ entries
* 6. Server prunes B's ops to MAX=20 while retaining the active import author
* 7. Client A receives B's pruned ops and applies them after its import
*
* IMPORTANT LIMITATION:
* With only 2 fresh E2E clients, vector clocks have ~3 entries — well below the
* MAX=20 pruning threshold. Without pruning, compareVectorClocks returns GREATER_THAN
* (correct!) even with the merge bug. The bug ONLY manifests when accumulated entries
* exceed MAX and pruning removes the import entry.
*
* SOLUTION: After Client B receives the SYNC_IMPORT, we inject 12 extra old-device
* entries into Client B's IndexedDB vector_clock store via page.evaluate(). This
* simulates real-world accumulated history from many past devices.
* TEST SETUP: We augment Client B's real upload with 25 old-device clock entries.
* This models a long-lived client and deliberately crosses the server's
* MAX_VECTOR_CLOCK_SIZE=20 boundary without manufacturing operations or responses.
* The historical counters deliberately outrank the import author. This proves
* the server preserves the active full-state boundary rather than merely keeping
* whichever entries have the highest counters.
*
* Run with: npm run e2e:supersync:file e2e/tests/sync/supersync-import-other-client-ops.spec.ts
*/
test.describe.configure({ mode: 'serial' });
interface StoredServerOperation {
id: string;
clientId: string;
vectorClock: Record<string, number>;
}
interface MutableUploadOperation {
id: string;
vectorClock: Record<string, number>;
}
interface MutableUploadBody {
ops: MutableUploadOperation[];
}
const getStoredOperations = async (
userId: number,
opType: string,
limit: number,
): Promise<StoredServerOperation[]> => {
const response = await fetch(
`${SUPERSYNC_BASE_URL}/api/test/user/${userId}/ops?opType=${opType}&limit=${limit}`,
);
if (!response.ok) {
throw new Error(
`Failed to inspect stored operation clocks: ${response.status} ${await response.text()}`,
);
}
const body = (await response.json()) as { ops: StoredServerOperation[] };
return body.ops;
};
test.describe('@supersync @pruning Other client post-import ops sync correctly', () => {
/**
* Scenario: Client B creates tasks after receiving Client A's SYNC_IMPORT
* with a bloated vector clock — tasks should sync to Client A
*
* This test injects old vector clock entries into Client B's IndexedDB to simulate
* real-world accumulated history. After pruning, the import's entry gets dropped,
* causing Client A to incorrectly filter Client B's ops as CONCURRENT.
* with an oversized vector clock — the server should prune it and the tasks
* should still sync to Client A.
*/
test('Other client post-import tasks with bloated clock sync to importing client', async ({
browser,
@ -156,110 +183,49 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
await waitForTask(clientB.page, 'E2E Import Test - Active Task With Subtask');
console.log('[Other-Client Import] Client B received SYNC_IMPORT');
// ============ PHASE 6: Inject bloated vector clock into Client B ============
// This simulates real-world accumulated history from many old devices.
// Without this injection, the vector clock only has ~3 entries (well below MAX=20)
// and pruning never happens, so the bug doesn't manifest.
const [fullStateOperation] = await getStoredOperations(
user.userId,
'SYNC_IMPORT',
1,
);
expect(fullStateOperation).toBeDefined();
const fullStateAuthor = fullStateOperation.clientId;
// ============ PHASE 6: Inflate Client B's real upload clock ============
console.log(
'[Other-Client Import] Phase 6: Injecting old device entries into Client B vector clock',
'[Other-Client Import] Phase 6: Installing oversized-clock upload route',
);
await clientB.page.evaluate(async () => {
interface VectorClockEntry {
clock: Record<string, number>;
lastUpdate: number;
const uploadedClockSizes: number[] = [];
const uploadedOperationIds: string[] = [];
await routeSuperSyncOps(clientB.page, async (route) => {
if (route.request().method() !== 'POST') {
await route.continue();
return;
}
const DB_NAME = 'SUP_OPS';
const VECTOR_CLOCK_STORE = 'vector_clock';
const SINGLETON_KEY = 'current';
// Open the database
const db = await new Promise<IDBDatabase>((resolve, reject) => {
const request = indexedDB.open(DB_NAME);
request.onsuccess = (): void => resolve(request.result);
request.onerror = (): void => reject(request.error);
});
// Read current vector clock entry
const currentEntry = await new Promise<VectorClockEntry | undefined>(
(resolve, reject) => {
const tx = db.transaction(VECTOR_CLOCK_STORE, 'readonly');
const store = tx.objectStore(VECTOR_CLOCK_STORE);
const request = store.get(SINGLETON_KEY);
request.onsuccess = (): void =>
resolve(request.result as VectorClockEntry | undefined);
request.onerror = (): void => reject(request.error);
},
);
const existingClock = currentEntry?.clock ?? {};
console.log(
'[Injected] Existing clock entries:',
Object.keys(existingClock).length,
);
// Add 12 old-device entries with high counters to simulate long history
// These represent old devices that were once part of the sync network
const bloatedClock = { ...existingClock };
for (let i = 0; i < 12; i++) {
const deviceId = `old-device-${String(i).padStart(5, '0')}`;
bloatedClock[deviceId] = 100 + i;
const body = parseSuperSyncRequestBody<MutableUploadBody>(route.request());
for (const operation of body.ops) {
for (let i = 0; i < 25; i++) {
operation.vectorClock[`old-device-${String(i).padStart(5, '0')}`] = 100 + i;
}
uploadedClockSizes.push(Object.keys(operation.vectorClock).length);
uploadedOperationIds.push(operation.id);
}
console.log(
'[Injected] Bloated clock entries:',
Object.keys(bloatedClock).length,
);
// Write back the bloated vector clock
await new Promise<void>((resolve, reject) => {
const tx = db.transaction(VECTOR_CLOCK_STORE, 'readwrite');
const store = tx.objectStore(VECTOR_CLOCK_STORE);
const entry = {
...currentEntry,
clock: bloatedClock,
lastUpdate: Date.now(),
};
const request = store.put(entry, SINGLETON_KEY);
request.onsuccess = (): void => resolve();
request.onerror = (): void => reject(request.error);
// Send the modified body as plain JSON even when the original browser
// request was gzip encoded.
const headers = { ...route.request().headers() };
delete headers['content-encoding'];
delete headers['content-transfer-encoding'];
delete headers['content-length'];
const response = await route.fetch({
headers,
postData: JSON.stringify(body),
});
db.close();
console.log('[Injected] Vector clock bloated successfully');
await route.fulfill({ response });
});
console.log(
'[Other-Client Import] Client B vector clock bloated with 12 old device entries',
);
// ============ PHASE 6b: Reload Client B to pick up injected clock ============
// CRITICAL: OperationLogStoreService has an in-memory _vectorClockCache that
// bypasses IndexedDB reads. Without reloading, the injected entries never reach
// the ops' vector clocks — getVectorClock() returns the stale cached value,
// and appendWithVectorClockOverwrite() overwrites the injected IDB data.
// Reloading destroys the Angular service, forcing a fresh read from IndexedDB.
console.log(
'[Other-Client Import] Phase 6b: Reloading Client B to pick up injected clock',
);
// Close the page and open a new one in the same context (preserves IndexedDB).
// Using page.reload() or page.goto() can hang when active sync connections
// prevent navigation lifecycle events from completing.
await clientB.page.close();
clientB.page = await clientB.context.newPage();
await clientB.page.goto('/');
await waitForAppReady(clientB.page, { ensureRoute: false });
// Re-create page objects for the new page
clientB.workView = new WorkViewPage(clientB.page, `B-${testRunId}`);
clientB.sync = new SuperSyncPage(clientB.page);
await clientB.sync.setupSuperSync(syncConfig);
await clientB.page.goto('/#/work-view');
await clientB.page.waitForLoadState('networkidle');
await waitForTask(clientB.page, 'E2E Import Test - Active Task With Subtask');
console.log('[Other-Client Import] Client B reloaded with bloated clock');
// ============ PHASE 7: Client B creates new tasks ============
// These ops will carry B's bloated vector clock with old entries
console.log('[Other-Client Import] Phase 7: Client B creating post-import tasks');
@ -276,7 +242,6 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
console.log(`[Other-Client Import] Client B created: ${taskB2}`);
// ============ PHASE 8: Client B syncs (uploads ops with post-import clock) ============
// Validates that B's ops (with clock from REPLACE, not MERGE) sync correctly
console.log(
'[Other-Client Import] Phase 8: Client B syncing (uploads ops with bloated clock)',
);
@ -284,6 +249,19 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
await clientB.sync.syncAndWait();
console.log('[Other-Client Import] Client B synced (ops uploaded)');
expect(uploadedClockSizes).toHaveLength(2);
expect(uploadedClockSizes.every((size) => size > 20)).toBe(true);
await unrouteSuperSyncOps(clientB.page);
const storedCreateOperations = (
await getStoredOperations(user.userId, 'CRT', 10)
).filter((operation) => uploadedOperationIds.includes(operation.id));
expect(storedCreateOperations).toHaveLength(2);
for (const operation of storedCreateOperations) {
expect(Object.keys(operation.vectorClock)).toHaveLength(20);
expect(operation.vectorClock[fullStateAuthor]).toBeDefined();
}
// ============ PHASE 9: Client A syncs (downloads B's pruned ops) ============
console.log('[Other-Client Import] Phase 9: Client A syncing (downloads B ops)');
@ -295,8 +273,6 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
await clientA.page.waitForLoadState('networkidle');
// ============ PHASE 10: Verify Client A sees B's tasks ============
// BUG: Client A filters B's ops as CONCURRENT with the import
// because B's pruned op clocks are missing the import's entry
console.log(
'[Other-Client Import] Phase 10: Verifying Client A has B post-import tasks',
);
@ -304,9 +280,7 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
// Both clients should have the imported task
await waitForTask(clientA.page, 'E2E Import Test - Active Task With Subtask');
// CRITICAL: Client A should have B's post-import tasks
// This is where the bug manifests — these tasks are missing on A
// because SyncImportFilterService filters them as CONCURRENT
// Client A should have B's post-import tasks after server-side pruning.
await waitForTask(clientA.page, taskB1);
await waitForTask(clientA.page, taskB2);
console.log('[Other-Client Import] Client A has B post-import tasks');
@ -331,7 +305,10 @@ test.describe('@supersync @pruning Other client post-import ops sync correctly',
console.log('[Other-Client Import] Other client post-import ops test PASSED!');
} finally {
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
if (clientB) {
await unrouteSuperSyncOps(clientB.page).catch(() => {});
await closeClient(clientB);
}
}
});
});

350
e2e/tests/sync/supersync-network-failure.spec.ts
View file

@ -4,6 +4,10 @@ import {
getSuperSyncConfig,
createSimulatedClient,
closeClient,
parseSuperSyncRequestBody,
routeSuperSyncOps,
SUPERSYNC_BASE_URL,
unrouteSuperSyncOps,
waitForTask,
type SimulatedE2EClient,
} from '../../utils/supersync-helpers';
@ -43,6 +47,7 @@ test.describe('@supersync Network Failure Recovery', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
let failNextUpload = true;
let failedUploadAttempts = 0;
try {
const user = await createTestUser(testRunId);
@ -57,9 +62,10 @@ test.describe('@supersync Network Failure Recovery', () => {
await clientA.workView.addTask(taskName);
// Set up route interception to fail first upload
await clientA.page.route('**/api/sync/ops/**', async (route) => {
await routeSuperSyncOps(clientA.page, async (route) => {
if (failNextUpload && route.request().method() === 'POST') {
failNextUpload = false;
failedUploadAttempts++;
console.log('[Test] Simulating upload failure');
await route.abort('failed');
} else {
@ -77,8 +83,10 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[Test] First sync failed as expected');
}
expect(failedUploadAttempts).toBe(1);
// Remove the failing route
await clientA.page.unroute('**/api/sync/ops/**');
await unrouteSuperSyncOps(clientA.page);
// Second sync attempt - should succeed
await clientA.sync.syncAndWait();
@ -99,6 +107,7 @@ test.describe('@supersync Network Failure Recovery', () => {
const taskLocatorB = clientB.page.locator(`task:has-text("${taskName}")`);
await expect(taskLocatorB).toBeVisible();
} finally {
if (clientA) await unrouteSuperSyncOps(clientA.page).catch(() => {});
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
}
@ -115,7 +124,7 @@ test.describe('@supersync Network Failure Recovery', () => {
test('recovers from download failure', async ({ browser, baseURL, testRunId }) => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
let failNextDownload = true;
let failedDownloadAttempts = 0;
try {
const user = await createTestUser(testRunId);
@ -132,20 +141,22 @@ test.describe('@supersync Network Failure Recovery', () => {
// Verify Client A has the task
await waitForTask(clientA.page, taskName);
// Set up Client B
// Install the failure before setup so automatic initial sync cannot race
// past the interception.
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
// Set up route interception to fail first download
await clientB.page.route('**/api/sync/ops/**', async (route) => {
if (failNextDownload && route.request().method() === 'GET') {
failNextDownload = false;
await routeSuperSyncOps(clientB.page, async (route) => {
if (route.request().method() === 'GET') {
failedDownloadAttempts++;
console.log('[Test] Simulating download failure');
await route.abort('failed');
} else {
await route.continue();
}
});
await clientB.sync.setupSuperSync({
...syncConfig,
waitForInitialSync: false,
});
// First sync attempt - download should fail
try {
@ -155,16 +166,14 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[Test] First download failed as expected');
}
expect(failedDownloadAttempts).toBeGreaterThan(0);
// Verify task NOT present (download failed)
const taskLocatorBeforeRetry = clientB.page.locator(`task:has-text("${taskName}")`);
await expect(taskLocatorBeforeRetry)
.not.toBeVisible({ timeout: 1000 })
.catch(() => {
// Task might or might not be visible depending on partial state
});
await expect(taskLocatorBeforeRetry).not.toBeVisible({ timeout: 1000 });
// Remove the failing route
await clientB.page.unroute('**/api/sync/ops/**');
await unrouteSuperSyncOps(clientB.page);
// Retry sync - should succeed
await clientB.sync.syncAndWait();
@ -175,7 +184,10 @@ test.describe('@supersync Network Failure Recovery', () => {
await expect(taskLocatorAfterRetry).toBeVisible();
} finally {
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
if (clientB) {
await unrouteSuperSyncOps(clientB.page).catch(() => {});
await closeClient(clientB);
}
}
});
@ -195,8 +207,7 @@ test.describe('@supersync Network Failure Recovery', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
// Use object to ensure mutable reference is captured correctly
const state = { returnServerError: true };
let serverErrorResponses = 0;
try {
const user = await createTestUser(testRunId);
@ -205,15 +216,11 @@ test.describe('@supersync Network Failure Recovery', () => {
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const taskName = `Task-${testRunId}-server-error`;
await clientA.workView.addTask(taskName);
// Wait for task to be fully created in store
await waitForTask(clientA.page, taskName);
// Intercept and return 500 error on first request
await clientA.page.route('**/api/sync/ops/**', async (route) => {
if (state.returnServerError && route.request().method() === 'POST') {
state.returnServerError = false;
// Intercept before creating the task so an automatic upload cannot race
// past the failure.
await routeSuperSyncOps(clientA.page, async (route) => {
if (route.request().method() === 'POST') {
serverErrorResponses++;
console.log('[Test] Simulating 500 server error');
await route.fulfill({
status: 500,
@ -225,6 +232,10 @@ test.describe('@supersync Network Failure Recovery', () => {
}
});
const taskName = `Task-${testRunId}-server-error`;
await clientA.workView.addTask(taskName);
await waitForTask(clientA.page, taskName);
// First sync - server error
try {
await clientA.sync.triggerSync();
@ -234,8 +245,10 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[Test] First sync got server error as expected');
}
expect(serverErrorResponses).toBeGreaterThan(0);
// Remove interception before retry
await clientA.page.unroute('**/api/sync/ops/**');
await unrouteSuperSyncOps(clientA.page);
// Give time for route to be fully removed
await clientA.page.waitForTimeout(500);
@ -254,7 +267,7 @@ test.describe('@supersync Network Failure Recovery', () => {
} finally {
// Ensure routes are cleaned up
if (clientA) {
await clientA.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
}
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
@ -355,18 +368,15 @@ test.describe('@supersync Network Failure Recovery', () => {
});
/**
* Test: Partial batch upload failure followed by successful retry
* Test: Server accepts an upload but the response is lost before local commit
*
* This tests a more realistic failure scenario where:
* 1. Client A creates 10 tasks rapidly
* 2. First sync starts - first few ops may succeed, then failure
* 3. Retry sync - all remaining ops should upload
* 4. Client B receives ALL 10 tasks (no duplicates, no missing)
*
* This verifies that the operation log correctly tracks which ops
* have been synced vs pending, and retry doesn't create duplicates.
* 2. Server commits the upload but the response is dropped
* 3. Client A reloads with the operations still locally pending
* 4. Retry deduplicates the already-stored operations
* 5. Client B receives all 10 tasks exactly once
*/
test('partial batch upload failure followed by retry uploads all without duplicates', async ({
test('accepted upload survives response loss and restart without duplicates', async ({
browser,
baseURL,
testRunId,
@ -375,10 +385,13 @@ test.describe('@supersync Network Failure Recovery', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
// Track how many POST requests we've seen
const state = {
requestCount: 0,
failAfter: 2, // Fail after 2 successful requests
phase: 'fault' as 'fault' | 'recovery',
committedUpload: false,
responseDropped: false,
acceptedOperationCount: 0,
committedOperationIds: [] as string[],
recoveryUploadIds: [] as string[][],
};
try {
@ -405,48 +418,88 @@ test.describe('@supersync Network Failure Recovery', () => {
await expect(taskLocator).toBeVisible();
}
// Set up route interception to fail after first few requests
await clientA.page.route('**/api/sync/ops/**', async (route) => {
// Let the first upload reach the real server, then hide its successful
// response from the client. Later attempts remain offline until reload.
await routeSuperSyncOps(clientA.page, async (route) => {
if (route.request().method() === 'POST') {
state.requestCount++;
if (state.requestCount > state.failAfter) {
console.log(
`[PartialBatch] Failing request #${state.requestCount} (after ${state.failAfter} successes)`,
);
const upload = parseSuperSyncRequestBody<{ ops: Array<{ id: string }> }>(
route.request(),
);
if (state.phase === 'fault') {
if (!state.committedUpload) {
state.committedOperationIds = upload.ops.map((operation) => operation.id);
const response = await route.fetch();
const body = (await response.json()) as {
results?: Array<{ accepted?: boolean }>;
};
state.acceptedOperationCount =
body.results?.filter((result) => result.accepted === true).length ?? 0;
state.committedUpload = true;
}
await route.abort('failed');
} else {
console.log(`[PartialBatch] Allowing request #${state.requestCount}`);
await route.continue();
state.responseDropped = true;
return;
}
state.recoveryUploadIds.push(upload.ops.map((operation) => operation.id));
await route.continue();
} else {
await route.continue();
}
});
// First sync attempt - will partially succeed then fail
console.log('[PartialBatch] Starting first sync (will partially fail)');
console.log('[ResponseLoss] Starting upload whose response will be dropped');
try {
await clientA.sync.triggerSync();
await clientA.page.waitForTimeout(3000);
} catch {
console.log('[PartialBatch] First sync failed as expected');
console.log('[ResponseLoss] Client observed the dropped upload response');
}
// Remove the failing route and reset counter
await clientA.page.unroute('**/api/sync/ops/**');
await clientA.page.waitForTimeout(500);
console.log('[PartialBatch] Route interception removed');
expect(state.responseDropped).toBe(true);
expect(state.committedUpload).toBe(true);
expect(state.committedOperationIds).toHaveLength(taskCount);
expect(state.acceptedOperationCount).toBe(taskCount);
// Retry sync - should succeed and upload remaining ops
console.log('[PartialBatch] Retrying sync');
const serverOpsAfterCommit = (await (
await fetch(`${SUPERSYNC_BASE_URL}/api/test/user/${user.userId}/ops?limit=100`)
).json()) as { ops: Array<{ id: string }> };
for (const operationId of state.committedOperationIds) {
expect(
serverOpsAfterCommit.ops.filter((operation) => operation.id === operationId),
).toHaveLength(1);
}
// Reload before acknowledging the accepted response. IndexedDB pending
// operations must retry idempotently after hydration.
state.phase = 'recovery';
await clientA.page.reload({ waitUntil: 'domcontentloaded' });
await clientA.workView.waitForTaskList();
console.log('[ResponseLoss] Retrying after reload');
await clientA.sync.syncAndWait();
console.log('[PartialBatch] Retry sync completed');
console.log('[ResponseLoss] Retry completed');
const committedIdSet = new Set(state.committedOperationIds);
expect(
state.recoveryUploadIds.some(
(ids) =>
ids.length === committedIdSet.size &&
ids.every((operationId) => committedIdSet.has(operationId)),
),
).toBe(true);
const serverOpsAfterRetry = (await (
await fetch(`${SUPERSYNC_BASE_URL}/api/test/user/${user.userId}/ops?limit=100`)
).json()) as { ops: Array<{ id: string }> };
for (const operationId of state.committedOperationIds) {
expect(
serverOpsAfterRetry.ops.filter((operation) => operation.id === operationId),
).toHaveLength(1);
}
// Verify all tasks still exist on Client A (no data loss)
for (const taskName of taskNames) {
await waitForTask(clientA.page, taskName);
}
console.log('[PartialBatch] All tasks still present on Client A');
console.log('[ResponseLoss] All tasks still present on Client A');
// Set up Client B
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
@ -466,14 +519,14 @@ test.describe('@supersync Network Failure Recovery', () => {
.count();
expect(countB).toBe(taskCount);
console.log(
`[PartialBatch] ✓ Client B has exactly ${taskCount} tasks (no duplicates)`,
`[ResponseLoss] ✓ Client B has exactly ${taskCount} tasks (no duplicates)`,
);
console.log('[PartialBatch] ✓ Partial batch failure + retry test PASSED!');
console.log('[ResponseLoss] ✓ Accepted upload retry remained idempotent');
} finally {
// Ensure routes are cleaned up
if (clientA) {
await clientA.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
}
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
@ -501,7 +554,7 @@ test.describe('@supersync Network Failure Recovery', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { returnRateLimitError: true };
let rateLimitResponses = 0;
try {
const user = await createTestUser(testRunId);
@ -510,14 +563,10 @@ test.describe('@supersync Network Failure Recovery', () => {
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const taskName = `Task-${testRunId}-rate-limit`;
await clientA.workView.addTask(taskName);
await waitForTask(clientA.page, taskName);
// Intercept and return 429 rate limit error on first request
await clientA.page.route('**/api/sync/ops/**', async (route) => {
if (state.returnRateLimitError && route.request().method() === 'POST') {
state.returnRateLimitError = false;
await routeSuperSyncOps(clientA.page, async (route) => {
if (rateLimitResponses === 0 && route.request().method() === 'POST') {
rateLimitResponses++;
console.log('[Test] Simulating 429 rate limit exceeded');
// eslint-disable-next-line @typescript-eslint/naming-convention
const headers = { 'Retry-After': '1' }; // Suggest retry after 1 second
@ -536,6 +585,10 @@ test.describe('@supersync Network Failure Recovery', () => {
}
});
const taskName = `Task-${testRunId}-rate-limit`;
await clientA.workView.addTask(taskName);
await waitForTask(clientA.page, taskName);
// First sync - rate limited
try {
await clientA.sync.triggerSync();
@ -544,8 +597,10 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[Test] First sync got rate limit error as expected');
}
expect(rateLimitResponses).toBe(1);
// Remove interception before retry
await clientA.page.unroute('**/api/sync/ops/**');
await unrouteSuperSyncOps(clientA.page);
await clientA.page.waitForTimeout(1500); // Wait longer than Retry-After
// Retry - should succeed now
@ -564,7 +619,7 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[RateLimit] ✓ Rate limit handling test PASSED');
} finally {
if (clientA) {
await clientA.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientA.page).catch(() => {});
}
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
@ -800,7 +855,7 @@ test.describe('@supersync Network Failure Recovery', () => {
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { returnMalformedJson: true };
let malformedResponses = 0;
try {
const user = await createTestUser(testRunId);
@ -815,14 +870,12 @@ test.describe('@supersync Network Failure Recovery', () => {
await clientA.sync.syncAndWait();
await waitForTask(clientA.page, taskName);
// Client B setup
// Install the malformed response before setup so initial auto-sync cannot
// download the task first.
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
// Intercept and return malformed JSON on first GET
await clientB.page.route('**/api/sync/ops/**', async (route) => {
if (state.returnMalformedJson && route.request().method() === 'GET') {
state.returnMalformedJson = false;
await routeSuperSyncOps(clientB.page, async (route) => {
if (route.request().method() === 'GET') {
malformedResponses++;
console.log('[Test] Simulating malformed JSON response');
await route.fulfill({
status: 200,
@ -833,6 +886,10 @@ test.describe('@supersync Network Failure Recovery', () => {
await route.continue();
}
});
await clientB.sync.setupSuperSync({
...syncConfig,
waitForInitialSync: false,
});
// First sync - should fail due to JSON parse error
try {
@ -842,8 +899,13 @@ test.describe('@supersync Network Failure Recovery', () => {
console.log('[Test] First sync failed due to malformed JSON as expected');
}
expect(malformedResponses).toBeGreaterThan(0);
await expect(
clientB.page.locator(`task:has-text("${taskName}")`),
).not.toBeVisible();
// Remove interception and retry
await clientB.page.unroute('**/api/sync/ops/**');
await unrouteSuperSyncOps(clientB.page);
await clientB.page.waitForTimeout(500);
// Retry - should succeed
@ -858,121 +920,7 @@ test.describe('@supersync Network Failure Recovery', () => {
} finally {
if (clientA) await closeClient(clientA);
if (clientB) {
await clientB.page.unroute('**/api/sync/ops/**').catch(() => {});
await closeClient(clientB);
}
}
});
/**
* Test: Invalid operation structure from server
*
* Scenario:
* 1. Client A creates a task
* 2. Server returns operations with missing required fields
* 3. Client should skip invalid operations and continue
* 4. Verify local state remains intact
*
* This tests that invalid/corrupted operations from the server
* are gracefully skipped without crashing the sync process.
*/
test('skips corrupted operations from server without crashing', async ({
browser,
baseURL,
testRunId,
}) => {
test.setTimeout(120000);
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
const state = { injectCorruptedOps: true };
try {
const user = await createTestUser(testRunId);
const syncConfig = getSuperSyncConfig(user);
// Client A creates and syncs a task
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const taskName = `Task-${testRunId}-corrupted-test`;
await clientA.workView.addTask(taskName);
await clientA.sync.syncAndWait();
// Setup Client B
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
// Intercept and inject corrupted operations
await clientB.page.route('**/api/sync/ops/**', async (route) => {
if (route.request().method() === 'GET' && state.injectCorruptedOps) {
state.injectCorruptedOps = false;
console.log('[Test] Injecting corrupted operations into response');
// Get real response then modify it
const response = await route.fetch();
const json = await response.json();
// Add corrupted operations to the response
if (json.ops) {
json.ops.push(
// Missing required fields
{
id: 'corrupt-1',
opType: 'UPD' /* missing entityType, payload, etc. */,
},
// Null payload
{
id: 'corrupt-2',
opType: 'UPD',
entityType: 'TASK',
entityId: 'bad-entity',
payload: null,
vectorClock: { bad: 1 },
timestamp: Date.now(),
},
// Invalid opType
{
id: 'corrupt-3',
opType: 'INVALID_TYPE',
entityType: 'TASK',
entityId: 'bad-entity',
payload: {},
vectorClock: { bad: 1 },
timestamp: Date.now(),
},
);
}
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify(json),
});
} else {
await route.continue();
}
});
// Sync - should handle corrupted ops gracefully
await clientB.sync.syncAndWait();
// Verify valid task was still received (corrupted ops didn't break sync)
await waitForTask(clientB.page, taskName);
const taskLocator = clientB.page.locator(`task:has-text("${taskName}")`);
await expect(taskLocator).toBeVisible();
// Remove interception
await clientB.page.unroute('**/api/sync/ops/**');
// Final sync should work normally
await clientB.sync.syncAndWait();
console.log('[CorruptedOps] ✓ Corrupted operation handling test PASSED');
} finally {
if (clientA) await closeClient(clientA);
if (clientB) {
await clientB.page.unroute('**/api/sync/ops/**').catch(() => {});
await unrouteSuperSyncOps(clientB.page).catch(() => {});
await closeClient(clientB);
}
}

7
e2e/tests/sync/supersync.spec.ts
View file

@ -314,6 +314,13 @@ test.describe('@supersync SuperSync E2E', () => {
* 6. Client B syncs
*
* Expected: Conflict detected or auto-merged, final state consistent
*
* WEAK — do not trust this as conflict coverage. Both clients change the SAME
* field (isDone), so there is no disjoint merge to get wrong, and the only
* assertion is that the task still exists. Strengthening it to rename-vs-done
* uncovered real data loss: the rename is committed as an op and then lost in
* sync. The stronger version lives in #9095 and stays out of this PR because it
* fails for a pre-existing reason unrelated to the fixes here.
*/
test('3.1 Concurrent edits handled gracefully', async ({
browser,

231
e2e/tests/sync/webdav-surgical-sync.spec.ts Normal file
View file

@ -0,0 +1,231 @@
import { test, expect } from '../../fixtures/webdav.fixture';
import type { APIRequestContext, Page } from '@playwright/test';
import { SyncPage } from '../../pages/sync.page';
import { WorkViewPage } from '../../pages/work-view.page';
import {
closeContextsSafely,
createSyncFolder,
generateSyncFolderName,
setupSyncClient,
waitForSyncComplete,
WEBDAV_CONFIG_TEMPLATE,
} from '../../utils/sync-helpers';
import { waitForAppReady, waitForStatePersistence } from '../../utils/waits';
interface SurgicalOpsFile {
recentOps: Array<{ id: string }>;
}
const readSurgicalOpsFile = async (
request: APIRequestContext,
url: string,
authorization: string,
): Promise<SurgicalOpsFile> => {
const response = await request.get(url, {
headers: { Authorization: authorization },
});
expect(response.ok()).toBe(true);
const encoded = await response.text();
const prefixEnd = encoded.indexOf('__');
if (prefixEnd < 0) {
throw new Error('Surgical ops file is missing its format prefix');
}
return JSON.parse(encoded.slice(prefixEnd + 2)) as SurgicalOpsFile;
};
const getLocalOperationState = async (
page: Page,
operationId: string,
): Promise<{ syncedAt?: number; rejectedAt?: number } | undefined> =>
page.evaluate(async (id) => {
const db = await new Promise<IDBDatabase>((resolve, reject) => {
const openRequest = indexedDB.open('SUP_OPS');
openRequest.onsuccess = (): void => resolve(openRequest.result);
openRequest.onerror = (): void => reject(openRequest.error);
});
try {
return await new Promise<{ syncedAt?: number; rejectedAt?: number } | undefined>(
(resolve, reject) => {
const tx = db.transaction('ops', 'readonly');
const getRequest = tx.objectStore('ops').index('byId').get(id);
getRequest.onsuccess = (): void =>
resolve(
getRequest.result as { syncedAt?: number; rejectedAt?: number } | undefined,
);
getRequest.onerror = (): void => reject(getRequest.error);
},
);
} finally {
db.close();
}
}, operationId);
test.describe('@webdav @surgical WebDAV Surgical sync', () => {
test('survives a committed ops response loss and restart', async ({
browser,
baseURL,
request,
}) => {
test.slow();
const appUrl = baseURL || 'http://localhost:4242';
const folderName = generateSyncFolderName('e2e-surgical');
const folderUrl = `${WEBDAV_CONFIG_TEMPLATE.baseUrl}${folderName}/DEV/`;
const config = {
...WEBDAV_CONFIG_TEMPLATE,
syncFolderPath: `/${folderName}`,
isUseSplitSyncFiles: true,
};
const authorization =
'Basic ' +
Buffer.from(
`${WEBDAV_CONFIG_TEMPLATE.username}:${WEBDAV_CONFIG_TEMPLATE.password}`,
).toString('base64');
await createSyncFolder(request, folderName);
let clientA: Awaited<ReturnType<typeof setupSyncClient>> | null = null;
let clientB: Awaited<ReturnType<typeof setupSyncClient>> | null = null;
try {
clientA = await setupSyncClient(browser, appUrl);
clientB = await setupSyncClient(browser, appUrl);
const syncA = new SyncPage(clientA.page);
const syncB = new SyncPage(clientB.page);
const workViewA = new WorkViewPage(clientA.page);
const workViewB = new WorkViewPage(clientB.page);
await syncA.setupWebdavSync(config);
const taskA = `Surgical-A-${folderName}`;
await workViewA.addTask(taskA);
await waitForStatePersistence(clientA.page);
await syncA.triggerSync();
await waitForSyncComplete(clientA.page, syncA);
const opsFile = await request.get(`${folderUrl}sync-ops.json`, {
headers: { Authorization: authorization },
});
const stateFile = await request.get(`${folderUrl}sync-state.json`, {
headers: { Authorization: authorization },
});
expect(opsFile.ok()).toBe(true);
expect(stateFile.ok()).toBe(true);
await syncB.setupWebdavSync(config);
await syncB.triggerSync();
await waitForSyncComplete(clientB.page, syncB);
await expect(clientB.page.locator('task', { hasText: taskA })).toBeVisible();
const baselineOps = await readSurgicalOpsFile(
request,
`${folderUrl}sync-ops.json`,
authorization,
);
const baselineOpIds = new Set(
baselineOps.recentOps.map((operation) => operation.id),
);
let requestPhase: 'fault' | 'restart' = 'fault';
const faultRequests: string[] = [];
const restartRequests: string[] = [];
let committedOpsWrite = false;
let responseDropped = false;
let restartOpsWrites = 0;
const recordWebDavRequest = (url: string): void => {
if (url.startsWith(folderUrl)) {
const requests = requestPhase === 'fault' ? faultRequests : restartRequests;
requests.push(new URL(url).pathname);
}
};
const requestListener = (webDavRequest: { url(): string }): void =>
recordWebDavRequest(webDavRequest.url());
clientB.page.on('request', requestListener);
await clientB.page.route('**/sync-ops.json', async (route) => {
if (route.request().method() === 'PUT') {
if (requestPhase === 'fault') {
if (!committedOpsWrite) {
const response = await route.fetch();
expect(response.ok()).toBe(true);
committedOpsWrite = true;
}
await route.abort('failed');
responseDropped = true;
return;
}
restartOpsWrites++;
}
await route.continue();
});
const taskB = `Surgical-B-${folderName}`;
await workViewB.addTask(taskB);
await waitForStatePersistence(clientB.page);
await syncB.triggerSync();
await expect.poll(() => responseDropped).toBe(true);
await syncB.syncSpinner.waitFor({ state: 'hidden', timeout: 20000 });
expect(committedOpsWrite).toBe(true);
expect(faultRequests.some((path) => path.endsWith('/sync-ops.json'))).toBe(true);
expect(faultRequests.some((path) => path.includes('/sync-state'))).toBe(false);
const committedOps = await readSurgicalOpsFile(
request,
`${folderUrl}sync-ops.json`,
authorization,
);
const newlyCommittedIds = committedOps.recentOps
.map((operation) => operation.id)
.filter((id) => !baselineOpIds.has(id));
expect(newlyCommittedIds).toHaveLength(1);
const committedOperationId = newlyCommittedIds[0];
expect(
(await getLocalOperationState(clientB.page, committedOperationId))?.syncedAt,
'the lost response must leave the committed operation pending',
).toBeUndefined();
// The ops PUT committed remotely, but its response never reached the
// client. Reload before retrying to exercise persisted cursor/revision
// recovery rather than only an in-memory retry.
requestPhase = 'restart';
await clientB.page.reload({ waitUntil: 'domcontentloaded' });
await waitForAppReady(clientB.page);
await syncB.triggerSync();
await waitForSyncComplete(clientB.page, syncB);
await expect
.poll(
async () =>
(await getLocalOperationState(clientB.page, committedOperationId))?.syncedAt,
)
.not.toBeUndefined();
await expect(clientB.page.locator('task', { hasText: taskB })).toHaveCount(1);
expect(restartRequests.some((path) => path.endsWith('/sync-ops.json'))).toBe(true);
expect(restartRequests.some((path) => path.includes('/sync-state'))).toBe(false);
expect(restartRequests.some((path) => path.endsWith('/sync-data.json'))).toBe(
false,
);
expect(restartOpsWrites).toBe(0);
const recoveredOps = await readSurgicalOpsFile(
request,
`${folderUrl}sync-ops.json`,
authorization,
);
expect(
recoveredOps.recentOps.filter(
(operation) => operation.id === committedOperationId,
),
).toHaveLength(1);
await syncA.triggerSync();
await waitForSyncComplete(clientA.page, syncA);
await expect(clientA.page.locator('task', { hasText: taskB })).toBeVisible();
await expect(clientB.page.locator('task', { hasText: taskA })).toBeVisible();
await expect(clientB.page.locator('task', { hasText: taskB })).toBeVisible();
clientB.page.off('request', requestListener);
} finally {
if (clientB) {
await clientB.page.unroute('**/sync-ops.json').catch(() => {});
}
await closeContextsSafely(clientA?.context, clientB?.context);
}
});
});

122
e2e/utils/supersync-helpers.ts
View file

@ -3,8 +3,11 @@ import {
type BrowserContext,
type Locator,
type Page,
type Request,
type Route,
expect,
} from '@playwright/test';
import { gunzipSync } from 'zlib';
import { SuperSyncPage, type SuperSyncConfig } from '../pages/supersync.page';
import { WorkViewPage } from '../pages/work-view.page';
import { waitForAppReady } from './waits';
@ -14,7 +17,6 @@ import {
UI_VISIBLE_TIMEOUT,
UI_VISIBLE_TIMEOUT_LONG,
UI_SETTLE_SMALL,
UI_SETTLE_MEDIUM,
UI_SETTLE_STANDARD,
UI_SETTLE_EXTENDED,
RETRY_BASE_DELAY,
@ -37,6 +39,37 @@ import {
export const SUPERSYNC_BASE_URL =
process.env.SUPERSYNC_E2E_URL || 'http://localhost:1901';
/**
* Matches both `/api/sync/ops` uploads and `/api/sync/ops?...` downloads.
* Do not add a trailing slash: the production endpoint has none.
*/
const SUPERSYNC_OPS_ROUTE = '**/api/sync/ops*';
export const routeSuperSyncOps = async (
page: Page,
handler: (route: Route) => Promise<void>,
): Promise<void> => page.route(SUPERSYNC_OPS_ROUTE, handler);
export const unrouteSuperSyncOps = async (page: Page): Promise<void> =>
page.unroute(SUPERSYNC_OPS_ROUTE);
/** Parse a SuperSync upload body regardless of whether the browser gzipped it. */
export const parseSuperSyncRequestBody = <T>(request: Request): T => {
try {
return request.postDataJSON() as T;
} catch {
const rawBody = request.postDataBuffer();
if (!rawBody) {
throw new Error('SuperSync request did not contain a body');
}
try {
return JSON.parse(gunzipSync(rawBody).toString('utf-8')) as T;
} catch (error) {
throw new Error('Failed to parse SuperSync request body', { cause: error });
}
}
};
/**
* Test user credentials returned from the server.
*/
@ -693,6 +726,14 @@ export const renameTask = async (
// Type directly into the textarea via evaluate to avoid focus/detach races
await textarea.evaluate((el: HTMLTextAreaElement, name: string) => {
el.focus();
// Dispatch focus explicitly rather than trusting el.focus() to emit it.
// These tests drive two clients as separate pages and only one page can hold
// focus, so on CI the event often never fires. TaskTitleComponent then keeps
// _isFocused=false, and its resetToLastExternalValueTrigger resets tmpValue
// to the stored title on the next task-object emission — blur therefore
// computes wasChanged=false, task.component skips update(), and the rename
// is silently dropped without ever becoming an op.
el.dispatchEvent(new Event('focus', { bubbles: true }));
el.value = name;
el.dispatchEvent(new Event('input', { bubbles: true }));
}, newName);
@ -700,7 +741,17 @@ export const renameTask = async (
await textarea.evaluate((el: HTMLTextAreaElement) => {
el.dispatchEvent(new Event('blur', { bubbles: true }));
});
await client.page.waitForTimeout(UI_SETTLE_MEDIUM);
// Assert against the STORE, not the DOM. task-title renders tmpValue (a
// component-local signal) in both its editing and idle branches, so a DOM
// check matches as soon as the synthetic input fires and can never tell a
// typed title from a committed one. Only the store proves an op was captured
// — and an uncaptured rename is invisible until a later sync reverts it.
await expect
.poll(() => getTaskTitleFromState(client, newName), {
timeout: UI_VISIBLE_TIMEOUT,
message: `renameTask: "${newName}" never reached the store — the rename was typed but never committed as an op`,
})
.toBe(newName);
};
/**
@ -836,6 +887,73 @@ export const waitForTaskTimeDisplay = async (
* @param taskName - The task name
* @returns The persisted timeSpent value in milliseconds, or null if not found
*/
/**
* Read a task's title from the live NgRx store.
*
* The DOM cannot answer this: task-title renders tmpValue, a component-local
* signal, in both its editing and idle branches, so it shows a typed title
* whether or not an op was ever captured. Only the store distinguishes them.
*
* @param client - The simulated E2E client
* @param titleSubstring - Substring identifying the task
* @returns The stored title, or null when no task matches
*/
export const getTaskTitleFromState = async (
client: SimulatedE2EClient,
titleSubstring: string,
): Promise<string | null> =>
client.page.evaluate(async (name) => {
const isRecord = (value: unknown): value is Record<string, unknown> =>
typeof value === 'object' && value !== null;
const getTitleFromRootState = (state: Record<string, unknown>): string | null => {
const taskState = state.tasks ?? state.task;
if (!isRecord(taskState) || !isRecord(taskState.entities)) {
return null;
}
for (const task of Object.values(taskState.entities)) {
if (
isRecord(task) &&
typeof task.title === 'string' &&
task.title.includes(name)
) {
return task.title;
}
}
return null;
};
type StoreSubscription = { unsubscribe: () => void };
type StoreLike = {
subscribe: (next: (state: unknown) => void) => StoreSubscription;
};
const helpers = (window as unknown as { __e2eTestHelpers?: { store?: StoreLike } })
.__e2eTestHelpers;
const store = helpers?.store;
if (!store) {
return null;
}
const liveState = await new Promise<Record<string, unknown> | null>((resolve) => {
let isDone = false;
const subscriptionRef: { current?: StoreSubscription } = {};
const finish = (state: unknown): void => {
if (isDone) {
return;
}
isDone = true;
window.setTimeout(() => subscriptionRef.current?.unsubscribe());
resolve(isRecord(state) ? state : null);
};
subscriptionRef.current = store.subscribe(finish);
window.setTimeout(() => finish(null), 1000);
});
return liveState ? getTitleFromRootState(liveState) : null;
}, titleSubstring);
export const getTaskTimeSpentFromState = async (
client: SimulatedE2EClient,
taskName: string,

32
packages/super-sync-server/src/sync/conflict.ts
View file

@ -288,7 +288,27 @@ export const isSameDuplicateOperation = (
maxClockDriftMs: number,
originalTimestamp: number = op.timestamp,
): boolean => {
const storedVectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]);
// Reproduce the normalization of the already-stored row. Storage may protect
// a low-counter causal full-state author in addition to the uploader, so use
// the stored clock's IDs as the authoritative protected set. Otherwise a
// genuine retry of an oversized post-import op compares as an ID collision.
//
// Tradeoff, deliberate: for an oversized incoming clock this projects onto the
// stored key set, so an id collision that ALSO differs only by an extra clock
// entry now reads as a duplicate instead of being caught. Accepted because
// every other structural field below (clientId, payload, entity, timestamp)
// must still match — and when they all do, acking beats rejecting a retry
// whose only sin is having learned about one more client.
const storedClockClientIds =
existingOp.vectorClock !== null &&
typeof existingOp.vectorClock === 'object' &&
!Array.isArray(existingOp.vectorClock)
? Object.keys(existingOp.vectorClock)
: [];
const storedVectorClock = limitVectorClockSize(op.vectorClock, [
op.clientId,
...storedClockClientIds,
]);
const incomingEncrypted = op.isPayloadEncrypted ?? false;
// encrypt() uses a fresh random IV per call, so retried ciphertext differs
@ -578,9 +598,15 @@ export const prefetchLatestEntityOpsForBatch = async (
return latestByEntity;
};
export const pruneVectorClockForStorage = (op: Operation): void => {
export const pruneVectorClockForStorage = (
op: Operation,
preserveClientIds: readonly string[] = [],
): void => {
const beforeSize = Object.keys(op.vectorClock).length;
op.vectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]);
op.vectorClock = limitVectorClockSize(op.vectorClock, [
op.clientId,
...preserveClientIds,
]);
const afterSize = Object.keys(op.vectorClock).length;
if (afterSize < beforeSize) {
Logger.debug(

80
packages/super-sync-server/src/sync/services/operation-upload.service.ts
View file

@ -8,6 +8,7 @@ import {
import {
AcceptedBatchOperation,
BatchUploadCandidate,
CAUSAL_FULL_STATE_OPERATION_WHERE,
CONFLICT_DETECTION_ENTITY_BATCH_SIZE,
ConflictResult,
DEFAULT_SYNC_CONFIG,
@ -15,6 +16,7 @@ import {
isCausalFullStateOperation,
isFullStateOpType,
limitVectorClockSize,
MAX_VECTOR_CLOCK_SIZE,
Operation,
OP_TYPES,
ProcessOperationResult,
@ -230,6 +232,69 @@ export class OperationUploadService {
});
}
/**
* Memoizes the causal full-state author per upload transaction. The answer can
* only change when THIS transaction accepts a causal full-state op, which
* {@link noteFullStateAuthor} folds back in — so one query per transaction is
* enough. Keyed by the `tx` client (a fresh short-lived object per
* `$transaction`), so entries cannot outlive the request that created them.
*/
private readonly fullStateAuthorByTx = new WeakMap<
Prisma.TransactionClient,
{ author: string | undefined }
>();
/**
* The latest causal full-state author remains a required clock edge for
* post-import operations. If pruning drops that low-counter entry, clients
* at the boundary classify the operation as concurrent and filter it out.
*
* Resolved lazily: only an op whose clock actually overflows pays for it.
*/
private async resolveFullStateAuthor(
tx: Prisma.TransactionClient,
userId: number,
): Promise<string | undefined> {
const memoized = this.fullStateAuthorByTx.get(tx);
if (memoized) {
return memoized.author;
}
const latestFullStateOp = await tx.operation.findFirst({
where: { userId, ...CAUSAL_FULL_STATE_OPERATION_WHERE },
orderBy: { serverSeq: 'desc' },
select: { clientId: true },
});
const author = latestFullStateOp?.clientId;
this.fullStateAuthorByTx.set(tx, { author });
return author;
}
/**
* Records a causal full-state op accepted earlier in this same transaction, so
* later ops protect the new author without re-reading (and without depending on
* read-your-writes visibility).
*/
private noteFullStateAuthor(tx: Prisma.TransactionClient, clientId: string): void {
this.fullStateAuthorByTx.set(tx, { author: clientId });
}
/**
* Protected clock IDs for storage: the uploader, plus the active causal
* full-state author when the clock is actually oversized. Under-limit clocks are
* never pruned, so they need no lookup at all.
*/
private async getPruneProtectedIds(
tx: Prisma.TransactionClient,
userId: number,
op: Operation,
): Promise<string[]> {
if (Object.keys(op.vectorClock).length <= MAX_VECTOR_CLOCK_SIZE) {
return [];
}
const author = await this.resolveFullStateAuthor(tx, userId);
return author ? [author] : [];
}
async processOperationBatch(
userId: number,
clientId: string,
@ -539,7 +604,11 @@ export class OperationUploadService {
}
}
pruneVectorClockForStorage(op);
if (isCausalFullStateOperation(op)) {
this.noteFullStateAuthor(tx, op.clientId);
}
const protectedIds = await this.getPruneProtectedIds(tx, userId, op);
pruneVectorClockForStorage(op, protectedIds);
// Reuse the payload byte size measured during validation; the clock is
// (re)measured inside computeOpStorageBytes because it was just pruned.
const sized = computeOpStorageBytes(op, candidate.payloadBytes);
@ -843,7 +912,14 @@ export class OperationUploadService {
// clock ID, causing the comparison to return CONCURRENT instead of GREATER_THAN,
// leading to an infinite rejection loop.
const beforeSize = Object.keys(op.vectorClock).length;
op.vectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]);
// Note this op's own authorship first: a causal full-state op is its own
// active author, so the memo answers without a query (and later ops in the
// same transaction see it).
if (isCausalFullStateOperation(op)) {
this.noteFullStateAuthor(tx, op.clientId);
}
const protectedIds = await this.getPruneProtectedIds(tx, userId, op);
op.vectorClock = limitVectorClockSize(op.vectorClock, [op.clientId, ...protectedIds]);
const afterSize = Object.keys(op.vectorClock).length;
if (afterSize < beforeSize) {
Logger.debug(

2
packages/super-sync-server/src/test-routes.ts
View file

@ -241,8 +241,10 @@ export const testRoutes = async (fastify: FastifyInstance): Promise<void> => {
take: limit,
select: {
id: true,
clientId: true,
opType: true,
serverSeq: true,
vectorClock: true,
},
});

20
packages/super-sync-server/tests/conflict.spec.ts
View file

@ -322,4 +322,24 @@ describe('conflict helpers', () => {
expect(Object.keys(incoming.vectorClock)).toHaveLength(MAX_VECTOR_CLOCK_SIZE);
expect(incoming.vectorClock['client-25']).toBe(25);
});
it('preserves the active full-state author while pruning', () => {
const fullStateAuthor = 'import-client';
const incoming = op({
clientId: 'upload-client',
vectorClock: {
[fullStateAuthor]: 1,
'upload-client': 2,
...Object.fromEntries(
Array.from({ length: 25 }, (_, index) => [`old-client-${index}`, 100 + index]),
),
},
});
pruneVectorClockForStorage(incoming, [fullStateAuthor]);
expect(Object.keys(incoming.vectorClock)).toHaveLength(MAX_VECTOR_CLOCK_SIZE);
expect(incoming.vectorClock[fullStateAuthor]).toBe(1);
expect(incoming.vectorClock['upload-client']).toBe(2);
});
});

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

@ -82,6 +82,19 @@ vi.mock('../src/db', async () => {
return { count };
}),
findFirst: vi.fn().mockImplementation(async (args: any) => {
// Shape of the full-state author query — counted so tests can pin that it
// stays one-per-upload rather than one-per-op. Selecting clientId ALONE is
// what separates it from the entity-conflict lookup, which also filters on
// OR + orders by serverSeq but selects the whole row.
if (
Array.isArray(args.where?.OR) &&
args.where?.entityType === undefined &&
args.orderBy?.serverSeq === 'desc' &&
args.select?.clientId === true &&
Object.keys(args.select).length === 1
) {
state.fullStateAuthorLookupCount++;
}
if (args.where?.id) {
return (
applyOperationSelect(state.operations.get(args.where.id), args.select) || null
@ -1012,6 +1025,124 @@ describe('SyncService', () => {
expect(testState.operations.size).toBe(25);
});
it.each([
['legacy serial', false],
['batch', true],
])(
'preserves the active full-state author when pruning in the %s path',
async (_label, batchUpload) => {
const service = new SyncService({ batchUpload });
const fullStateAuthor = 'import-author';
const uploadClient = 'post-import-client';
const fullStateOp = makeOp({
clientId: fullStateAuthor,
actionType: '[SP_ALL] Load(import) all data',
opType: 'SYNC_IMPORT',
entityType: 'ALL',
entityId: undefined,
payload: { TASK: {} },
vectorClock: { [fullStateAuthor]: 1 },
});
const oversizedDelta = makeOp({
clientId: uploadClient,
entityId: 'post-import-task',
vectorClock: {
[fullStateAuthor]: 1,
[uploadClient]: 2,
...Object.fromEntries(
Array.from({ length: 25 }, (_, index) => [
`old-client-${index}`,
100 + index,
]),
),
},
timestamp: fullStateOp.timestamp + 1,
});
const retryDelta = makeOp({
...oversizedDelta,
vectorClock: { ...oversizedDelta.vectorClock },
});
expect(
(await service.uploadOps(userId, fullStateAuthor, [fullStateOp]))[0].accepted,
).toBe(true);
expect(
(await service.uploadOps(userId, uploadClient, [oversizedDelta]))[0].accepted,
).toBe(true);
const storedClock = testState.operations.get(oversizedDelta.id)?.vectorClock as
| Record<string, number>
| undefined;
expect(storedClock).toBeDefined();
expect(Object.keys(storedClock ?? {})).toHaveLength(20);
expect(storedClock?.[fullStateAuthor]).toBe(1);
expect(storedClock?.[uploadClient]).toBe(2);
expect((await service.uploadOps(userId, uploadClient, [retryDelta]))[0]).toEqual(
expect.objectContaining({
accepted: false,
errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION,
}),
);
},
);
it.each([
['legacy serial', false],
['batch', true],
])(
'looks the full-state author up at most once per upload in the %s path',
async (_label, batchUpload) => {
// The answer cannot change mid-transaction unless this upload itself
// accepts a full-state op, so one oversized op must not become one query.
const service = new SyncService({ batchUpload });
const fullStateAuthor = 'import-author';
const uploadClient = 'post-import-client';
const fullStateOp = makeOp({
clientId: fullStateAuthor,
actionType: '[SP_ALL] Load(import) all data',
opType: 'SYNC_IMPORT',
entityType: 'ALL',
entityId: undefined,
payload: { TASK: {} },
vectorClock: { [fullStateAuthor]: 1 },
});
expect(
(await service.uploadOps(userId, fullStateAuthor, [fullStateOp]))[0].accepted,
).toBe(true);
const oversizedDeltas = Array.from({ length: 5 }, (_, index) =>
makeOp({
clientId: uploadClient,
entityId: `post-import-task-${index}`,
vectorClock: {
[fullStateAuthor]: 1,
[uploadClient]: 2 + index,
...Object.fromEntries(
Array.from({ length: 25 }, (_, old) => [`old-client-${old}`, 100 + old]),
),
},
timestamp: fullStateOp.timestamp + 1 + index,
}),
);
testState.fullStateAuthorLookupCount = 0;
const results = await service.uploadOps(userId, uploadClient, oversizedDeltas);
expect(results.every(({ accepted }) => accepted)).toBe(true);
expect(testState.fullStateAuthorLookupCount).toBe(1);
// The saved query must not cost the protection it exists for.
for (const delta of oversizedDeltas) {
const storedClock = testState.operations.get(delta.id)?.vectorClock as
| Record<string, number>
| undefined;
expect(Object.keys(storedClock ?? {})).toHaveLength(20);
expect(storedClock?.[fullStateAuthor]).toBe(1);
}
},
);
it.each([
['legacy serial', false],
['batch', true],

2
packages/super-sync-server/tests/sync.service.test-state.ts
View file

@ -12,6 +12,7 @@ export const testState = {
serverSeqCounter: 0,
batchConflictQueryCount: 0,
entityConflictFindFirstCount: 0,
fullStateAuthorLookupCount: 0,
};
export function resetTestState(): void {
@ -22,6 +23,7 @@ export function resetTestState(): void {
testState.serverSeqCounter = 0;
testState.batchConflictQueryCount = 0;
testState.entityConflictFindFirstCount = 0;
testState.fullStateAuthorLookupCount = 0;
}
export function applyOperationSelect(op: any, select?: Record<string, boolean>): any {

2
src/app/imex/sync/sync-config.service.spec.ts
View file

@ -87,6 +87,7 @@ describe('SyncConfigService', () => {
syncProvider: SyncProviderId.WebDAV,
syncInterval: 300000,
isEncryptionEnabled: true,
isUseSplitSyncFiles: true,
encryptKey: 'secret-key',
webDav: {
baseUrl: 'https://example.com',
@ -104,6 +105,7 @@ describe('SyncConfigService', () => {
expect(globalConfigService.updateSection).toHaveBeenCalledWith('sync', {
isEnabled: true,
isEncryptionEnabled: true,
isUseSplitSyncFiles: true,
syncProvider: SyncProviderId.WebDAV,
syncInterval: 300000,
});

3
src/app/imex/sync/sync-config.service.ts
View file

@ -342,6 +342,9 @@ export class SyncConfigService {
...(newSettings.isManualSyncOnly !== undefined
? { isManualSyncOnly: newSettings.isManualSyncOnly }
: {}),
...(newSettings.isUseSplitSyncFiles !== undefined
? { isUseSplitSyncFiles: newSettings.isUseSplitSyncFiles }
: {}),
};
// Provider-specific settings (URLs, credentials) must be stored securely
if (providerId) {

39
src/app/op-log/sync-providers/file-based/file-based-sync-adapter.service.spec.ts
View file

@ -3321,6 +3321,45 @@ describe('FileBasedSyncAdapterService', () => {
});
});
it('(a) acknowledges an already-committed op retry without appending it again', async () => {
const opsFile = makeOpsFile({
syncVersion: 3,
recentOps: [makeCompactOp({ id: 'op-123', sv: 3 })],
});
routeDownloads({ [C.OPS_FILE]: addPrefix(opsFile, 3) });
const result = await adapter.uploadOps([createMockSyncOp()], 'client1');
expect(result).toEqual({
results: [{ opId: 'op-123', accepted: true }],
latestSeq: 3,
});
expect(uploadedPaths()).toEqual([]);
});
it('(a) still heals a corrupt primary when every op is already in the recovered .bak', async () => {
// .bak recovery hands back an ops file that ALREADY contains the pending op
// (its PUT committed, then the response was lost). The all-duplicates
// short-circuit must NOT fire here: this cycle's conditional overwrite is
// the only thing that repairs the corrupt primary.
const recovered = makeOpsFile({
syncVersion: 3,
recentOps: [makeCompactOp({ id: 'op-123', sv: 3 })],
});
mockProvider.downloadFile.and.callFake(async (path: string) => {
if (path === C.OPS_BACKUP_FILE) {
return { dataStr: addPrefix(recovered, 3), rev: 'ops-bak-rev' };
}
throw new SyncDataCorruptedError('corrupt', C.OPS_FILE);
});
// Seeds the sync-cycle cache with the CORRUPT primary's rev.
await adapter.downloadOps(0);
await adapter.uploadOps([createMockSyncOp()], 'client1');
expect(uploadedPaths()).toContain(C.OPS_FILE);
});
it('(a) op-only download reads ONLY sync-ops.json (no sync-state.json fetch)', async () => {
const opsFile = makeOpsFile({
syncVersion: 5,

82
src/app/op-log/sync-providers/file-based/file-based-sync-adapter.service.ts
View file

@ -215,7 +215,15 @@ export class FileBasedSyncAdapterService {
*/
private _splitOpsCache = new Map<
string,
{ data: FileBasedOpsFile; rev: string; timestamp: number }
{
data: FileBasedOpsFile;
rev: string;
timestamp: number;
// True when `data` came from .bak recovery and `rev` is the CORRUPT
// primary's rev. This cycle's upload MUST still perform its conditional
// overwrite to heal the primary — see _stageOrDropDownloadedRev.
isPrimaryCorrupt: boolean;
}
>();
/** Tracks whether we've loaded persisted state from localStorage */
@ -1595,22 +1603,32 @@ export class FileBasedSyncAdapterService {
private _getCachedOpsData(
providerKey: string,
): { data: FileBasedOpsFile; rev: string } | null {
): { data: FileBasedOpsFile; rev: string; isPrimaryCorrupt: boolean } | null {
const cached = this._splitOpsCache.get(providerKey);
if (!cached) return null;
if (Date.now() - cached.timestamp > this._CACHE_TTL_MS) {
this._splitOpsCache.delete(providerKey);
return null;
}
return { data: cached.data, rev: cached.rev };
return {
data: cached.data,
rev: cached.rev,
isPrimaryCorrupt: cached.isPrimaryCorrupt,
};
}
private _setCachedOpsData(
providerKey: string,
data: FileBasedOpsFile,
rev: string,
isPrimaryCorrupt = false,
): void {
this._splitOpsCache.set(providerKey, { data, rev, timestamp: Date.now() });
this._splitOpsCache.set(providerKey, {
data,
rev,
timestamp: Date.now(),
isPrimaryCorrupt,
});
}
private _clearCachedOpsData(providerKey: string): void {
@ -2263,11 +2281,16 @@ export class FileBasedSyncAdapterService {
): Promise<OpUploadResponse> {
let opsFile: FileBasedOpsFile | null = null;
let opsRev: string | null = null;
// When the cached ops file came from .bak recovery, the primary is corrupt and
// only a conditional overwrite from this cycle heals it. Never short-circuit
// that write, however little there is to append.
let isPrimaryCorrupt = false;
const cached = this._getCachedOpsData(providerKey);
if (cached) {
opsFile = cached.data;
opsRev = cached.rev;
isPrimaryCorrupt = cached.isPrimaryCorrupt;
} else {
try {
const r = await this._downloadOpsFile(provider, cfg, encryptKey);
@ -2308,21 +2331,44 @@ export class FileBasedSyncAdapterService {
return { results: [], latestSeq: this._localSeqCounters.get(providerKey) || 0 };
}
const existingOps: SyncFileCompactOp[] = opsFile?.recentOps || [];
const existingOpIndexById = new Map(
existingOps.map((operation, index) => [operation.id, index]),
);
const newOps = ops.filter((operation) => !existingOpIndexById.has(operation.id));
// A successful PUT may commit even when its response is lost. The local op
// then remains pending and is retried after restart. Acknowledge IDs already
// present in the remote buffer without advancing syncVersion or appending a
// second copy.
//
// Skipped while the primary is corrupt: the ops below were read from .bak, and
// this cycle's conditional overwrite is what heals the primary. Falling through
// rewrites the file (appending nothing) and repairs it.
if (newOps.length === 0 && opsFile && !isPrimaryCorrupt) {
// serverSeq is omitted (it is optional): an op already in the buffer was
// numbered by a previous upload, and inventing a value here would mix two
// numbering schemes. The caller acknowledges on `opId` + `accepted`.
return {
results: ops.map((operation) => ({ opId: operation.id, accepted: true })),
latestSeq: opsFile.syncVersion,
};
}
const currentSyncVersion = opsFile?.syncVersion ?? 0;
const newSyncVersion = currentSyncVersion + 1;
const compactOps: SyncFileCompactOp[] = ops.map((op) => ({
const compactOps: SyncFileCompactOp[] = newOps.map((op) => ({
...this._syncOpToCompact(op),
sv: newSyncVersion,
}));
let mergedClock: VectorClock = opsFile?.vectorClock || {};
for (const op of ops) {
for (const op of newOps) {
mergedClock = mergeVectorClocks(mergedClock, op.vectorClock);
}
const schemaVersion = ops[0]?.schemaVersion || opsFile?.schemaVersion || 1;
const schemaVersion = newOps[0]?.schemaVersion || opsFile?.schemaVersion || 1;
const existingOps: SyncFileCompactOp[] = opsFile?.recentOps || [];
const combinedOps = [...existingOps, ...compactOps];
// Compaction is required when the folder has no snapshot yet (fresh/first
@ -2465,13 +2511,19 @@ export class FileBasedSyncAdapterService {
this._persistState();
const latestSeq = newSyncVersion;
const startingSeq = Math.max(0, latestSeq - ops.length);
const startingSeq = Math.max(0, latestSeq - newOps.length);
let newOperationIndex = 0;
return {
results: ops.map((op, i) => ({
opId: op.id,
accepted: true,
serverSeq: startingSeq + i + 1,
})),
// Already-committed ops are acknowledged without a serverSeq — see the
// all-duplicates return above.
results: ops.map((op) => {
if (existingOpIndexById.has(op.id)) {
return { opId: op.id, accepted: true };
}
const serverSeq = startingSeq + newOperationIndex + 1;
newOperationIndex++;
return { opId: op.id, accepted: true, serverSeq };
}),
latestSeq,
};
}
@ -2556,7 +2608,7 @@ export class FileBasedSyncAdapterService {
);
opsFile = recovered.data;
opsRev = (e as { primaryRev?: string } | null)?.primaryRev ?? recovered.rev;
this._setCachedOpsData(providerKey, opsFile, opsRev);
this._setCachedOpsData(providerKey, opsFile, opsRev, true);
this._notifyRecoveredCorruptPrimaryOnce(providerKey, opsRev);
} else {
throw e;