mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
* fix(sync): persist lastServerSeq after piggybacked ops applied #8304 The upload path persisted lastServerSeq covering piggybacked ops before those ops were applied (processRemoteOps runs in the caller, after the upload lock releases). A crash or a cancelled SYNC_IMPORT dialog in that window advanced the seq past unstored ops, so the next download skipped them forever. The upload service now defers the seq persist to the caller once it collects piggybacked ops; the orchestrator persists only after processRemoteOps succeeds, and skips it on the cancel/USE_REMOTE/USE_LOCAL early-returns. The non-piggyback path still persists in-loop (no loss risk). Mirrors the download path's 'persist AFTER ops stored' invariant. Updates 3 upload specs that encoded the buggy persist-before-apply and adds upload-side ordering + cancel-path regression specs. * test(sync): add crash-window regression for deferred lastServerSeq #8304 * test(sync): cross-service integration test for piggyback seq persistence #8304 Wires the real OperationLogUploadService + real OperationLogSyncService over one in-memory provider seq. Verifies the seq is not advanced on cancel/crash and only advanced after processRemoteOps on success. Fails against pre-fix code (all 3). * test(sync): e2e guard for cancel-reconvergence of incoming SYNC_IMPORT Multi-client SuperSync spec covering a scenario no existing test does: an incoming SYNC_IMPORT conflicting with a LOCAL PENDING op -> conflict dialog -> CANCEL -> the next sync RE-OFFERS it -> USE_REMOTE reconverges. Documents that it routes through the (already-correct) download path, so it is a real-server guard for the shared cancel->re-offer->reconverge contract, NOT a #8304 fix regression (that path is race-only and covered by the unit + cross-service integration specs). Requires docker.
This commit is contained in:
Johannes Millan
GitHub
parent
5c6c29dad8
commit
56334897cc
7 changed files with 792 additions and 18 deletions
|
|
@ -591,4 +591,142 @@ test.describe('@supersync @import-conflict Sync Import Conflict Dialog', () => {
|
|||
if (clientB) await closeClient(clientB);
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Reconvergence guard: an incoming SYNC_IMPORT that conflicts with a LOCAL PENDING op
|
||||
* shows the conflict dialog; cancelling it must leave the client able to recover —
|
||||
* the next sync RE-OFFERS the import and the user can adopt it (USE_REMOTE) to
|
||||
* reconverge. No existing spec covers this scenario (the CANCEL test above is the
|
||||
* local-import path; the silent-accept test below has no pending op).
|
||||
*
|
||||
* SCOPE / what this does and does NOT verify (related to #8304, but not a fix
|
||||
* regression for it): the app syncs download-first, so in this deterministic flow the
|
||||
* incoming SYNC_IMPORT is intercepted by Client B's DOWNLOAD phase, which already
|
||||
* persisted lastServerSeq correctly before #8304 — so this test passes with or
|
||||
* without that fix. #8304's actual bug is on the UPLOAD-piggyback path, reachable only
|
||||
* via a timing race (A uploads between B's download and upload) that a deterministic
|
||||
* E2E cannot force; that path is covered by the unit specs and
|
||||
* operation-log-upload-piggyback-seq.integration.spec.ts. This test is the real-server
|
||||
* end-to-end guard for the shared cancel→re-offer→reconverge contract.
|
||||
*
|
||||
* Actions:
|
||||
* 1. Clients A and B sync a shared baseline task.
|
||||
* 2. Client B creates a NEW local task and does NOT sync it (meaningful pending op).
|
||||
* 3. Client A imports a backup → SYNC_IMPORT lands on the server.
|
||||
* 4. Client B syncs → conflict dialog → CANCEL.
|
||||
* 5. Client B syncs AGAIN → conflict dialog appears AGAIN (re-offered, not lost).
|
||||
* 6. Client B chooses USE_REMOTE → reconverges to the imported state.
|
||||
*/
|
||||
test('CANCEL of an incoming SYNC_IMPORT (with a local pending op) re-offers it and reconverges', async ({
|
||||
browser,
|
||||
baseURL,
|
||||
testRunId,
|
||||
}) => {
|
||||
test.slow();
|
||||
|
||||
const uniqueId = Date.now();
|
||||
let clientA: SimulatedE2EClient | null = null;
|
||||
let clientB: SimulatedE2EClient | null = null;
|
||||
|
||||
try {
|
||||
const user = await createTestUser(testRunId);
|
||||
const syncConfig = getSuperSyncConfig(user);
|
||||
|
||||
// ============ PHASE 1: Both clients sync a shared baseline task ============
|
||||
console.log('[#8304] Phase 1: Both clients sync a shared baseline task');
|
||||
|
||||
clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId);
|
||||
await clientA.sync.setupSuperSync(syncConfig);
|
||||
|
||||
clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId);
|
||||
await clientB.sync.setupSuperSync(syncConfig);
|
||||
|
||||
const sharedTask = `Shared-Task-${uniqueId}`;
|
||||
await clientA.workView.addTask(sharedTask);
|
||||
await clientA.sync.syncAndWait();
|
||||
await clientB.sync.syncAndWait();
|
||||
await waitForTask(clientB.page, sharedTask);
|
||||
console.log('[#8304] Both clients synced shared baseline task');
|
||||
|
||||
// ============ PHASE 2: Client B creates a pending (unsynced) local op ============
|
||||
console.log('[#8304] Phase 2: Client B creates a pending local task (no sync)');
|
||||
|
||||
const bPendingTask = `B-Pending-${uniqueId}`;
|
||||
await clientB.workView.addTask(bPendingTask);
|
||||
await waitForTask(clientB.page, bPendingTask);
|
||||
// Deliberately do NOT sync B — this leaves a meaningful pending op so the
|
||||
// incoming SYNC_IMPORT produces a conflict dialog rather than silent accept.
|
||||
|
||||
// ============ PHASE 3: Client A imports a backup and pushes it ============
|
||||
console.log('[#8304] Phase 3: Client A imports backup and pushes SYNC_IMPORT');
|
||||
|
||||
const importPage = new ImportPage(clientA.page);
|
||||
await importPage.navigateToImportPage();
|
||||
const backupPath = ImportPage.getFixturePath('test-backup.json');
|
||||
await importPage.importBackupFile(backupPath);
|
||||
await clientA.page.goto(clientA.page.url(), {
|
||||
waitUntil: 'domcontentloaded',
|
||||
timeout: 30000,
|
||||
});
|
||||
await clientA.page.waitForLoadState('networkidle');
|
||||
// A's local SYNC_IMPORT conflicts with A's previously-synced ops; USE_LOCAL pushes.
|
||||
await clientA.sync.syncAndWait({ useLocal: true });
|
||||
console.log('[#8304] Client A pushed SYNC_IMPORT to server');
|
||||
|
||||
// ============ PHASE 4: Client B syncs and CANCELS the conflict dialog ============
|
||||
console.log('[#8304] Phase 4: Client B syncs → conflict dialog → CANCEL');
|
||||
|
||||
// Click sync directly (not syncAndWait) so the dialog is not auto-dismissed.
|
||||
await clientB.sync.syncBtn.click();
|
||||
const dialog = clientB.page.locator('dialog-sync-import-conflict');
|
||||
await expect(dialog).toBeVisible({ timeout: 20000 });
|
||||
await dialog.getByRole('button', { name: /cancel/i }).click();
|
||||
await expect(dialog).not.toBeVisible({ timeout: 5000 });
|
||||
console.log('[#8304] Client B cancelled the dialog');
|
||||
|
||||
// ============ PHASE 5: B unchanged — still has its pending + baseline, no import ============
|
||||
console.log('[#8304] Phase 5: Verify Client B state is unchanged after cancel');
|
||||
|
||||
await clientB.page.goto('/#/work-view');
|
||||
await clientB.page.waitForLoadState('networkidle');
|
||||
await waitForTask(clientB.page, bPendingTask);
|
||||
await waitForTask(clientB.page, sharedTask);
|
||||
const importedOnBAfterCancel = clientB.page.locator(
|
||||
'task:has-text("E2E Import Test - Active Task With Subtask")',
|
||||
);
|
||||
await expect(importedOnBAfterCancel).not.toBeVisible({ timeout: 5000 });
|
||||
console.log('[#8304] ✓ Client B unchanged (import not applied, pending op intact)');
|
||||
|
||||
// ============ PHASE 6: #8304 KEY — next sync RE-OFFERS the SYNC_IMPORT ============
|
||||
console.log('[#8304] Phase 6: Client B syncs again → dialog must re-appear');
|
||||
|
||||
// If a cancel had advanced the server cursor past A's SYNC_IMPORT (the class of
|
||||
// defect #8304 fixes on the upload path), the next download would skip it and NO
|
||||
// dialog would appear — Client B would be permanently diverged. The dialog
|
||||
// re-appearing proves the import was re-fetched.
|
||||
await clientB.sync.syncBtn.click();
|
||||
await expect(dialog).toBeVisible({ timeout: 20000 });
|
||||
console.log('[#8304] ✓ SYNC_IMPORT was re-offered on the next sync (not lost)');
|
||||
|
||||
// ============ PHASE 7: Adopt server data → reconverge ============
|
||||
console.log('[#8304] Phase 7: Client B chooses USE_REMOTE to reconverge');
|
||||
|
||||
await dialog.getByRole('button', { name: /server/i }).click();
|
||||
await expect(dialog).not.toBeVisible({ timeout: 10000 });
|
||||
await clientB.sync.waitForSyncToComplete();
|
||||
|
||||
await clientB.page.goto('/#/work-view');
|
||||
await clientB.page.waitForLoadState('networkidle');
|
||||
await waitForTask(clientB.page, 'E2E Import Test - Active Task With Subtask');
|
||||
// USE_REMOTE discards B's local-only pending op in favour of the imported state.
|
||||
const pendingAfterAdopt = clientB.page.locator(`task:has-text("${bPendingTask}")`);
|
||||
await expect(pendingAfterAdopt).not.toBeVisible({ timeout: 5000 });
|
||||
console.log('[#8304] ✓ Client B reconverged on the imported state');
|
||||
|
||||
console.log('[#8304] ✓ Cancel-reconvergence test PASSED!');
|
||||
} finally {
|
||||
if (clientA) await closeClient(clientA);
|
||||
if (clientB) await closeClient(clientB);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -139,6 +139,18 @@ export interface UploadResult {
|
|||
* Callers should skip post-upload logic (LWW re-upload, IN_SYNC status).
|
||||
*/
|
||||
cancelled?: boolean;
|
||||
/**
|
||||
* The lastServerSeq value the caller must persist via setLastServerSeq AFTER it has
|
||||
* applied the piggybacked ops (processRemoteOps). Only set when piggybacked ops were
|
||||
* collected for the caller to apply; undefined otherwise (the upload service persisted
|
||||
* the seq itself, since advancing past our own uploaded ops carries no loss risk).
|
||||
*
|
||||
* Deferring the persist mirrors the download path's invariant ("persist lastServerSeq
|
||||
* AFTER ops are stored"): if a crash or a cancelled SYNC_IMPORT dialog occurs between
|
||||
* upload return and processRemoteOps, the seq must NOT have advanced past those ops,
|
||||
* or the next download skips them forever. (#8304)
|
||||
*/
|
||||
lastServerSeqToPersist?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -402,6 +402,172 @@ describe('OperationLogSyncService', () => {
|
|||
).toHaveBeenCalledWith(jasmine.objectContaining({ isNeverSynced: true }));
|
||||
});
|
||||
|
||||
// #8304: the upload service defers persisting lastServerSeq for piggybacked ops;
|
||||
// the orchestrator must persist it ONLY after processRemoteOps applies them.
|
||||
describe('lastServerSeq persistence for piggybacked ops (#8304)', () => {
|
||||
it('should persist lastServerSeq AFTER processing piggybacked ops', async () => {
|
||||
opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([]));
|
||||
|
||||
const piggybackedOp: Operation = {
|
||||
id: 'piggybacked-1',
|
||||
clientId: 'client-B',
|
||||
actionType: 'test' as ActionType,
|
||||
opType: OpType.Update,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-1',
|
||||
payload: { title: 'Remote Title' },
|
||||
vectorClock: { clientB: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
|
||||
uploadServiceSpy.uploadPendingOps.and.resolveTo({
|
||||
uploadedCount: 1,
|
||||
piggybackedOps: [piggybackedOp],
|
||||
rejectedCount: 0,
|
||||
rejectedOps: [],
|
||||
lastServerSeqToPersist: 77,
|
||||
});
|
||||
|
||||
// Track order: setLastServerSeq must run AFTER processRemoteOps.
|
||||
const callOrder: string[] = [];
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.callFake(async () => {
|
||||
callOrder.push('processRemoteOps');
|
||||
return {
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
};
|
||||
});
|
||||
const setLastServerSeqSpy = jasmine
|
||||
.createSpy('setLastServerSeq')
|
||||
.and.callFake(async () => {
|
||||
callOrder.push('setLastServerSeq');
|
||||
});
|
||||
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: setLastServerSeqSpy,
|
||||
} as any;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider);
|
||||
|
||||
expect(result.kind).toBe('completed');
|
||||
expect(setLastServerSeqSpy).toHaveBeenCalledWith(77);
|
||||
// The seq must NOT advance before the ops it covers are applied.
|
||||
expect(callOrder).toEqual(['processRemoteOps', 'setLastServerSeq']);
|
||||
});
|
||||
|
||||
it('should NOT persist lastServerSeq when a piggybacked SYNC_IMPORT dialog is cancelled', async () => {
|
||||
// A meaningful local op remains pending so the gate produces dialog data for
|
||||
// the incoming SYNC_IMPORT, which the user then cancels (default dialog result).
|
||||
opLogStoreSpy.hasSyncedOps.and.resolveTo(false);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([
|
||||
{
|
||||
seq: 1,
|
||||
op: {
|
||||
id: 'local-task-create',
|
||||
clientId: 'client-A',
|
||||
actionType: 'test' as ActionType,
|
||||
opType: OpType.Create,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-1',
|
||||
payload: { title: 'Real local task' },
|
||||
vectorClock: { clientA: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
},
|
||||
appliedAt: Date.now(),
|
||||
source: 'local',
|
||||
},
|
||||
]);
|
||||
|
||||
const piggybackedSyncImport: Operation = {
|
||||
id: 'remote-sync-import',
|
||||
clientId: 'client-B',
|
||||
actionType: ActionType.LOAD_ALL_DATA,
|
||||
opType: OpType.SyncImport,
|
||||
entityType: 'ALL',
|
||||
payload: {},
|
||||
vectorClock: { clientB: 5 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
|
||||
uploadServiceSpy.uploadPendingOps.and.resolveTo({
|
||||
uploadedCount: 0,
|
||||
piggybackedOps: [piggybackedSyncImport],
|
||||
rejectedCount: 0,
|
||||
rejectedOps: [],
|
||||
lastServerSeqToPersist: 99,
|
||||
});
|
||||
|
||||
const setLastServerSeqSpy = jasmine
|
||||
.createSpy('setLastServerSeq')
|
||||
.and.resolveTo();
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: setLastServerSeqSpy,
|
||||
} as any;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider);
|
||||
|
||||
expect(result.kind).toBe('cancelled');
|
||||
// CRITICAL: the seq must NOT advance — the piggybacked ops (the SYNC_IMPORT and
|
||||
// any siblings) were never applied, so the next download must re-fetch them.
|
||||
expect(setLastServerSeqSpy).not.toHaveBeenCalled();
|
||||
expect(remoteOpsProcessingServiceSpy.processRemoteOps).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should NOT persist lastServerSeq if processRemoteOps throws (crash window)', async () => {
|
||||
opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([]));
|
||||
|
||||
const piggybackedOp: Operation = {
|
||||
id: 'piggybacked-1',
|
||||
clientId: 'client-B',
|
||||
actionType: 'test' as ActionType,
|
||||
opType: OpType.Update,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-1',
|
||||
payload: { title: 'Test' },
|
||||
vectorClock: { clientB: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
|
||||
uploadServiceSpy.uploadPendingOps.and.resolveTo({
|
||||
uploadedCount: 1,
|
||||
piggybackedOps: [piggybackedOp],
|
||||
rejectedCount: 0,
|
||||
rejectedOps: [],
|
||||
lastServerSeqToPersist: 88,
|
||||
});
|
||||
|
||||
// Simulate a crash mid-apply (e.g. DecryptNoPasswordError, validation throw).
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.rejectWith(
|
||||
new Error('apply failed mid-flight'),
|
||||
);
|
||||
|
||||
const setLastServerSeqSpy = jasmine
|
||||
.createSpy('setLastServerSeq')
|
||||
.and.resolveTo();
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: setLastServerSeqSpy,
|
||||
} as any;
|
||||
|
||||
await expectAsync(service.uploadPendingOps(mockProvider)).toBeRejected();
|
||||
|
||||
// The seq must NOT have advanced — the apply threw before completing, so the
|
||||
// next download must re-fetch the piggybacked ops instead of skipping them.
|
||||
expect(setLastServerSeqSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('rejected ops handling delegation', () => {
|
||||
let mockProvider: any;
|
||||
|
||||
|
|
|
|||
|
|
@ -298,6 +298,15 @@ export class OperationLogSyncService {
|
|||
);
|
||||
localWinOpsCreated = processResult.localWinOpsCreated;
|
||||
// Validation failure (if any) is on the session-validation latch.
|
||||
|
||||
// #8304: Persist lastServerSeq ONLY now that the piggybacked ops have been applied
|
||||
// above. The upload service deferred this (see UploadResult.lastServerSeqToPersist)
|
||||
// so that a crash — or a cancelled/USE_REMOTE/USE_LOCAL SYNC_IMPORT dialog, all of
|
||||
// which return early ABOVE without reaching here — cannot advance the seq past ops
|
||||
// that were never stored. Mirrors the download path's invariant.
|
||||
if (result.lastServerSeqToPersist !== undefined) {
|
||||
await syncProvider.setLastServerSeq(result.lastServerSeqToPersist);
|
||||
}
|
||||
}
|
||||
|
||||
// STEP 2: Handle server-rejected operations
|
||||
|
|
|
|||
|
|
@ -0,0 +1,410 @@
|
|||
import { TestBed } from '@angular/core/testing';
|
||||
import { provideMockStore } from '@ngrx/store/testing';
|
||||
import { OperationLogSyncService } from './operation-log-sync.service';
|
||||
import { OperationLogUploadService } from './operation-log-upload.service';
|
||||
import { OperationEncryptionService } from './operation-encryption.service';
|
||||
import { OperationLogStoreService } from '../persistence/operation-log-store.service';
|
||||
import { SchemaMigrationService } from '../persistence/schema-migration.service';
|
||||
import { SnackService } from '../../core/snack/snack.service';
|
||||
import { VectorClockService } from './vector-clock.service';
|
||||
import { OperationApplierService } from '../apply/operation-applier.service';
|
||||
import { ConflictResolutionService } from './conflict-resolution.service';
|
||||
import { ValidateStateService } from '../validation/validate-state.service';
|
||||
import { RepairOperationService } from '../validation/repair-operation.service';
|
||||
import { OperationLogDownloadService } from './operation-log-download.service';
|
||||
import { LockService } from './lock.service';
|
||||
import { OperationLogCompactionService } from '../persistence/operation-log-compaction.service';
|
||||
import { SyncImportFilterService } from './sync-import-filter.service';
|
||||
import { ServerMigrationService } from './server-migration.service';
|
||||
import { SupersededOperationResolverService } from './superseded-operation-resolver.service';
|
||||
import { RemoteOpsProcessingService } from './remote-ops-processing.service';
|
||||
import { RejectedOpsHandlerService } from './rejected-ops-handler.service';
|
||||
import { OperationWriteFlushService } from './operation-write-flush.service';
|
||||
import { SuperSyncStatusService } from './super-sync-status.service';
|
||||
import { SyncHydrationService } from '../persistence/sync-hydration.service';
|
||||
import { SyncImportConflictDialogService } from './sync-import-conflict-dialog.service';
|
||||
import { StateSnapshotService } from '../backup/state-snapshot.service';
|
||||
import { BackupService } from '../backup/backup.service';
|
||||
import { TranslateService } from '@ngx-translate/core';
|
||||
import { ActionType, OperationLogEntry, OpType } from '../core/operation.types';
|
||||
import { INBOX_PROJECT } from '../../features/project/project.const';
|
||||
import { TODAY_TAG } from '../../features/tag/tag.const';
|
||||
import type {
|
||||
OperationSyncCapable,
|
||||
SyncOperation,
|
||||
} from '../sync-providers/provider.interface';
|
||||
|
||||
/**
|
||||
* #8304 cross-service integration test.
|
||||
*
|
||||
* Unlike the unit specs (which mock OperationLogUploadService out of
|
||||
* OperationLogSyncService), this wires the REAL OperationLogUploadService and the
|
||||
* REAL OperationLogSyncService.uploadPendingOps together over a single fake provider
|
||||
* whose lastServerSeq lives in one in-memory variable. That round-trip is the part
|
||||
* the unit specs cannot exercise: the upload service must DEFER the seq persist and
|
||||
* the sync service must persist it ONLY after the piggybacked ops are applied.
|
||||
*
|
||||
* The invariant under test (the data-loss window #8304 closes): the persisted
|
||||
* lastServerSeq must NEVER advance past piggybacked ops that were not applied —
|
||||
* whether the user cancels the SYNC_IMPORT dialog or the apply step throws. On
|
||||
* success it must advance, but only after processRemoteOps completes.
|
||||
*
|
||||
* processRemoteOps and the conflict-gate's downstream apply are mocked (controllable
|
||||
* crash/cancel); everything between provider.uploadOps and provider.setLastServerSeq
|
||||
* is the real two-service code path.
|
||||
*/
|
||||
describe('OperationLogSyncService + OperationLogUploadService — piggyback seq persistence (#8304)', () => {
|
||||
let service: OperationLogSyncService;
|
||||
let opLogStoreSpy: jasmine.SpyObj<OperationLogStoreService>;
|
||||
let remoteOpsProcessingServiceSpy: jasmine.SpyObj<RemoteOpsProcessingService>;
|
||||
let dialogServiceSpy: jasmine.SpyObj<SyncImportConflictDialogService>;
|
||||
|
||||
// The single source of truth for the server cursor, shared by both real services
|
||||
// through the fake provider. Assertions read this directly.
|
||||
let persistedServerSeq: number;
|
||||
let setLastServerSeqSpy: jasmine.Spy;
|
||||
let uploadOpsSpy: jasmine.Spy;
|
||||
|
||||
const INITIAL_SEQ = 40;
|
||||
const SERVER_LATEST_SEQ = 50;
|
||||
|
||||
// A meaningful local pending op so the conflict gate produces dialog data when a
|
||||
// SYNC_IMPORT is piggybacked (mirrors the existing unit spec's gate setup).
|
||||
const localPendingEntry: OperationLogEntry = {
|
||||
seq: 1,
|
||||
op: {
|
||||
id: 'local-task-create',
|
||||
clientId: 'client-A',
|
||||
actionType: 'test' as ActionType,
|
||||
opType: OpType.Create,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-1',
|
||||
payload: { title: 'Real local task' },
|
||||
vectorClock: { clientA: 1 },
|
||||
timestamp: 1700000000000,
|
||||
schemaVersion: 1,
|
||||
},
|
||||
appliedAt: 1700000000000,
|
||||
source: 'local',
|
||||
};
|
||||
|
||||
const makePiggybackSyncImport = (): SyncOperation => ({
|
||||
id: 'remote-sync-import',
|
||||
clientId: 'client-B',
|
||||
actionType: ActionType.LOAD_ALL_DATA,
|
||||
opType: OpType.SyncImport,
|
||||
entityType: 'ALL',
|
||||
payload: {},
|
||||
vectorClock: { clientB: 5 },
|
||||
timestamp: 1700000000001,
|
||||
schemaVersion: 1,
|
||||
});
|
||||
|
||||
const makePiggybackRegularUpdate = (): SyncOperation => ({
|
||||
id: 'remote-update',
|
||||
clientId: 'client-B',
|
||||
actionType: '[Task] Update',
|
||||
opType: OpType.Update,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-9',
|
||||
payload: { changes: { title: 'Remote update' } },
|
||||
vectorClock: { clientB: 5 },
|
||||
timestamp: 1700000000001,
|
||||
schemaVersion: 1,
|
||||
});
|
||||
|
||||
/**
|
||||
* Builds a fake OperationSyncCapable provider whose uploadOps piggybacks the given
|
||||
* ops and whose lastServerSeq is backed by `persistedServerSeq`.
|
||||
*/
|
||||
const makeProvider = (piggybackOps: SyncOperation[]): OperationSyncCapable => {
|
||||
uploadOpsSpy = jasmine.createSpy('uploadOps').and.callFake(async () => ({
|
||||
results: [{ opId: localPendingEntry.op.id, accepted: true }],
|
||||
latestSeq: SERVER_LATEST_SEQ,
|
||||
newOps: piggybackOps.map((op, i) => ({
|
||||
serverSeq: SERVER_LATEST_SEQ - (piggybackOps.length - 1 - i),
|
||||
receivedAt: 1700000000002,
|
||||
op,
|
||||
})),
|
||||
}));
|
||||
setLastServerSeqSpy = jasmine
|
||||
.createSpy('setLastServerSeq')
|
||||
.and.callFake(async (n: number) => {
|
||||
persistedServerSeq = n;
|
||||
});
|
||||
return {
|
||||
isReady: () => Promise.resolve(true),
|
||||
supportsOperationSync: true,
|
||||
getLastServerSeq: () => Promise.resolve(persistedServerSeq),
|
||||
setLastServerSeq: setLastServerSeqSpy,
|
||||
uploadOps: uploadOpsSpy,
|
||||
// No getEncryptKey → encryption disabled → no decrypt/encrypt plumbing needed.
|
||||
} as unknown as OperationSyncCapable;
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
persistedServerSeq = INITIAL_SEQ;
|
||||
|
||||
opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [
|
||||
'getUnsynced',
|
||||
'markSynced',
|
||||
'markRejected',
|
||||
'loadStateCache',
|
||||
'getLastSeq',
|
||||
'getOpById',
|
||||
'setVectorClock',
|
||||
'clearFullStateOps',
|
||||
'getVectorClock',
|
||||
'appendBatchSkipDuplicates',
|
||||
'hasSyncedOps',
|
||||
'deleteOpsWhere',
|
||||
]);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([localPendingEntry]);
|
||||
opLogStoreSpy.markSynced.and.resolveTo(undefined);
|
||||
opLogStoreSpy.markRejected.and.resolveTo(undefined);
|
||||
opLogStoreSpy.setVectorClock.and.resolveTo();
|
||||
opLogStoreSpy.clearFullStateOps.and.resolveTo();
|
||||
opLogStoreSpy.getVectorClock.and.resolveTo(null);
|
||||
opLogStoreSpy.deleteOpsWhere.and.resolveTo();
|
||||
opLogStoreSpy.appendBatchSkipDuplicates.and.resolveTo({
|
||||
seqs: [],
|
||||
writtenOps: [],
|
||||
skippedCount: 0,
|
||||
});
|
||||
// Never-synced snapshot is captured before upload; false ⇒ gate treats a USE_LOCAL
|
||||
// choice as potentially overwriting remote, so it produces dialog data.
|
||||
opLogStoreSpy.hasSyncedOps.and.resolveTo(false);
|
||||
// Not a wholly fresh client (so upload is not blocked).
|
||||
opLogStoreSpy.loadStateCache.and.resolveTo({
|
||||
state: {},
|
||||
lastAppliedOpSeq: 1,
|
||||
vectorClock: {},
|
||||
compactedAt: 1700000000000,
|
||||
});
|
||||
opLogStoreSpy.getLastSeq.and.resolveTo(1);
|
||||
|
||||
remoteOpsProcessingServiceSpy = jasmine.createSpyObj('RemoteOpsProcessingService', [
|
||||
'processRemoteOps',
|
||||
]);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
});
|
||||
|
||||
dialogServiceSpy = jasmine.createSpyObj('SyncImportConflictDialogService', [
|
||||
'showConflictDialog',
|
||||
]);
|
||||
dialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL');
|
||||
|
||||
const lockServiceSpy = jasmine.createSpyObj('LockService', ['request']);
|
||||
// The real upload service runs its body inside lockService.request(name, cb).
|
||||
lockServiceSpy.request.and.callFake((_name: string, cb: () => Promise<unknown>) =>
|
||||
cb(),
|
||||
);
|
||||
|
||||
const serverMigrationServiceSpy = jasmine.createSpyObj('ServerMigrationService', [
|
||||
'checkAndHandleMigration',
|
||||
'handleServerMigration',
|
||||
]);
|
||||
serverMigrationServiceSpy.checkAndHandleMigration.and.resolveTo();
|
||||
serverMigrationServiceSpy.handleServerMigration.and.resolveTo();
|
||||
|
||||
const stateSnapshotServiceSpy = jasmine.createSpyObj('StateSnapshotService', [
|
||||
'getStateSnapshot',
|
||||
]);
|
||||
stateSnapshotServiceSpy.getStateSnapshot.and.returnValue({
|
||||
task: { ids: [] },
|
||||
project: { ids: [INBOX_PROJECT.id] },
|
||||
tag: { ids: [TODAY_TAG.id] },
|
||||
note: { ids: [] },
|
||||
} as any);
|
||||
|
||||
const rejectedOpsHandlerServiceSpy = jasmine.createSpyObj(
|
||||
'RejectedOpsHandlerService',
|
||||
['handleRejectedOps'],
|
||||
);
|
||||
rejectedOpsHandlerServiceSpy.handleRejectedOps.and.resolveTo({
|
||||
mergedOpsCreated: 0,
|
||||
permanentRejectionCount: 0,
|
||||
});
|
||||
|
||||
const writeFlushServiceSpy = jasmine.createSpyObj('OperationWriteFlushService', [
|
||||
'flushPendingWrites',
|
||||
]);
|
||||
writeFlushServiceSpy.flushPendingWrites.and.resolveTo();
|
||||
|
||||
const superSyncStatusServiceSpy = jasmine.createSpyObj('SuperSyncStatusService', [
|
||||
'updatePendingOpsStatus',
|
||||
]);
|
||||
|
||||
const encryptionServiceSpy = jasmine.createSpyObj('OperationEncryptionService', [
|
||||
'encryptOperations',
|
||||
'decryptOperations',
|
||||
'encryptPayload',
|
||||
]);
|
||||
|
||||
TestBed.configureTestingModule({
|
||||
providers: [
|
||||
OperationLogSyncService,
|
||||
// REAL upload service — this is the point of the integration test.
|
||||
OperationLogUploadService,
|
||||
provideMockStore(),
|
||||
{ provide: OperationEncryptionService, useValue: encryptionServiceSpy },
|
||||
{ provide: OperationLogStoreService, useValue: opLogStoreSpy },
|
||||
{ provide: LockService, useValue: lockServiceSpy },
|
||||
{ provide: RemoteOpsProcessingService, useValue: remoteOpsProcessingServiceSpy },
|
||||
{ provide: SyncImportConflictDialogService, useValue: dialogServiceSpy },
|
||||
{ provide: ServerMigrationService, useValue: serverMigrationServiceSpy },
|
||||
{ provide: StateSnapshotService, useValue: stateSnapshotServiceSpy },
|
||||
{ provide: RejectedOpsHandlerService, useValue: rejectedOpsHandlerServiceSpy },
|
||||
{ provide: OperationWriteFlushService, useValue: writeFlushServiceSpy },
|
||||
{ provide: SuperSyncStatusService, useValue: superSyncStatusServiceSpy },
|
||||
{
|
||||
provide: SnackService,
|
||||
useValue: jasmine.createSpyObj('SnackService', ['open']),
|
||||
},
|
||||
{
|
||||
provide: BackupService,
|
||||
useValue: jasmine.createSpyObj('BackupService', [
|
||||
'captureImportBackup',
|
||||
'restoreImportBackup',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: SchemaMigrationService,
|
||||
useValue: jasmine.createSpyObj('SchemaMigrationService', [
|
||||
'getCurrentVersion',
|
||||
'migrateOperation',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: VectorClockService,
|
||||
useValue: jasmine.createSpyObj('VectorClockService', [
|
||||
'getEntityFrontier',
|
||||
'getSnapshotVectorClock',
|
||||
'getSnapshotEntityKeys',
|
||||
'getCurrentVectorClock',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: OperationApplierService,
|
||||
useValue: jasmine.createSpyObj('OperationApplierService', ['applyOperations']),
|
||||
},
|
||||
{
|
||||
provide: ConflictResolutionService,
|
||||
useValue: jasmine.createSpyObj('ConflictResolutionService', [
|
||||
'autoResolveConflictsLWW',
|
||||
'checkOpForConflicts',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: ValidateStateService,
|
||||
useValue: jasmine.createSpyObj('ValidateStateService', [
|
||||
'validateAndRepair',
|
||||
'validateAndRepairCurrentState',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: RepairOperationService,
|
||||
useValue: jasmine.createSpyObj('RepairOperationService', [
|
||||
'createRepairOperation',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: OperationLogDownloadService,
|
||||
useValue: jasmine.createSpyObj('OperationLogDownloadService', [
|
||||
'downloadRemoteOps',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: OperationLogCompactionService,
|
||||
useValue: jasmine.createSpyObj('OperationLogCompactionService', ['compact']),
|
||||
},
|
||||
{
|
||||
provide: TranslateService,
|
||||
useValue: jasmine.createSpyObj('TranslateService', ['instant']),
|
||||
},
|
||||
{
|
||||
provide: SyncImportFilterService,
|
||||
useValue: jasmine.createSpyObj('SyncImportFilterService', [
|
||||
'filterOpsInvalidatedBySyncImport',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: SupersededOperationResolverService,
|
||||
useValue: jasmine.createSpyObj('SupersededOperationResolverService', [
|
||||
'resolveSupersededLocalOps',
|
||||
]),
|
||||
},
|
||||
{
|
||||
provide: SyncHydrationService,
|
||||
useValue: jasmine.createSpyObj('SyncHydrationService', [
|
||||
'hydrateFromRemoteSync',
|
||||
]),
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
service = TestBed.inject(OperationLogSyncService);
|
||||
});
|
||||
|
||||
it('does NOT advance the persisted server seq when the piggybacked SYNC_IMPORT dialog is cancelled', async () => {
|
||||
const provider = makeProvider([makePiggybackSyncImport()]);
|
||||
|
||||
const result = await service.uploadPendingOps(provider);
|
||||
|
||||
expect(result.kind).toBe('cancelled');
|
||||
// Upload actually piggybacked ops back (the real upload service ran).
|
||||
expect(uploadOpsSpy).toHaveBeenCalled();
|
||||
// CRITICAL (#8304): the cursor stayed at its pre-sync value, so the next download
|
||||
// re-fetches the SYNC_IMPORT (and siblings) instead of skipping them forever.
|
||||
expect(persistedServerSeq).toBe(INITIAL_SEQ);
|
||||
expect(setLastServerSeqSpy).not.toHaveBeenCalled();
|
||||
// The dialog was cancelled before any apply happened.
|
||||
expect(remoteOpsProcessingServiceSpy.processRemoteOps).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does NOT advance the persisted server seq when applying piggybacked ops throws (crash window)', async () => {
|
||||
// Regular piggybacked update (no SYNC_IMPORT ⇒ no dialog), then the apply throws.
|
||||
const provider = makeProvider([makePiggybackRegularUpdate()]);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.rejectWith(
|
||||
new Error('apply failed mid-flight'),
|
||||
);
|
||||
|
||||
await expectAsync(service.uploadPendingOps(provider)).toBeRejected();
|
||||
|
||||
// CRITICAL (#8304): a crash between upload-return and apply must leave the cursor
|
||||
// untouched so the ops are re-downloaded, not skipped.
|
||||
expect(persistedServerSeq).toBe(INITIAL_SEQ);
|
||||
expect(setLastServerSeqSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('advances the persisted server seq AFTER piggybacked ops are applied (success)', async () => {
|
||||
const provider = makeProvider([makePiggybackRegularUpdate()]);
|
||||
|
||||
// Record ordering to prove the persist happens strictly after the apply.
|
||||
const callOrder: string[] = [];
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.callFake(async () => {
|
||||
callOrder.push('processRemoteOps');
|
||||
return {
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
};
|
||||
});
|
||||
setLastServerSeqSpy.and.callFake(async (n: number) => {
|
||||
callOrder.push('setLastServerSeq');
|
||||
persistedServerSeq = n;
|
||||
});
|
||||
|
||||
const result = await service.uploadPendingOps(provider);
|
||||
|
||||
expect(result.kind).toBe('completed');
|
||||
// Cursor advanced to the server's latest only after the apply completed.
|
||||
expect(persistedServerSeq).toBe(SERVER_LATEST_SEQ);
|
||||
expect(callOrder).toEqual(['processRemoteOps', 'setLastServerSeq']);
|
||||
});
|
||||
});
|
||||
|
|
@ -406,8 +406,11 @@ describe('OperationLogUploadService', () => {
|
|||
|
||||
const result = await service.uploadPendingOps(mockApiProvider);
|
||||
|
||||
// Should use max serverSeq from piggybacked ops (50), not latestSeq (100)
|
||||
expect(mockApiProvider.setLastServerSeq).toHaveBeenCalledWith(50);
|
||||
// #8304: piggybacked ops were collected for the caller to apply, so the seq
|
||||
// persist is DEFERRED to the caller (no in-loop persist). The deferred value
|
||||
// is the max serverSeq from piggybacked ops (50), not latestSeq (100).
|
||||
expect(result.lastServerSeqToPersist).toBe(50);
|
||||
expect(mockApiProvider.setLastServerSeq).not.toHaveBeenCalled();
|
||||
expect(result.hasMorePiggyback).toBe(true);
|
||||
});
|
||||
|
||||
|
|
@ -461,13 +464,11 @@ describe('OperationLogUploadService', () => {
|
|||
|
||||
const result = await service.uploadPendingOps(mockApiProvider);
|
||||
|
||||
// Verify setLastServerSeq calls
|
||||
const calls = mockApiProvider.setLastServerSeq.calls.allArgs();
|
||||
expect(calls.length).toBe(2);
|
||||
// First chunk: should store 100 (latestSeq, since no hasMorePiggyback)
|
||||
expect(calls[0][0]).toBe(100);
|
||||
// Second chunk: should NOT regress to 50, should keep 100
|
||||
expect(calls[1][0]).toBe(100);
|
||||
// #8304: chunk 1 collected piggybacked ops, so the seq persist is deferred to
|
||||
// the caller for ALL subsequent chunks (no in-loop persist). The deferred value
|
||||
// must be the highest non-regressing seq (100), never regressing to chunk 2's 50.
|
||||
expect(mockApiProvider.setLastServerSeq).not.toHaveBeenCalled();
|
||||
expect(result.lastServerSeqToPersist).toBe(100);
|
||||
expect(result.hasMorePiggyback).toBe(true);
|
||||
});
|
||||
|
||||
|
|
@ -551,15 +552,35 @@ describe('OperationLogUploadService', () => {
|
|||
}
|
||||
});
|
||||
|
||||
await service.uploadPendingOps(mockApiProvider);
|
||||
const result = await service.uploadPendingOps(mockApiProvider);
|
||||
|
||||
// Verify setLastServerSeq calls
|
||||
const calls = mockApiProvider.setLastServerSeq.calls.allArgs();
|
||||
expect(calls.length).toBe(2);
|
||||
// First chunk: should store 55 (max of piggybacked ops)
|
||||
expect(calls[0][0]).toBe(55);
|
||||
// Second chunk: should keep 55 (Math.max(55, 45) = 55), not regress to 45
|
||||
expect(calls[1][0]).toBe(55);
|
||||
// #8304: both chunks collected piggybacked ops, so the seq persist is deferred
|
||||
// to the caller. The deferred value tracks the highest received seq across
|
||||
// chunks (max(55, 45) = 55), never regressing to chunk 2's 45.
|
||||
expect(mockApiProvider.setLastServerSeq).not.toHaveBeenCalled();
|
||||
expect(result.lastServerSeqToPersist).toBe(55);
|
||||
});
|
||||
|
||||
// #8304 regression: when a chunk receives NO piggybacked ops, the seq only
|
||||
// covers our own just-uploaded ops, so persisting in-loop carries no loss risk
|
||||
// and the caller has nothing to persist afterwards.
|
||||
it('should persist seq in-loop (not defer) when no piggybacked ops are received', async () => {
|
||||
mockApiProvider.getLastServerSeq.and.returnValue(Promise.resolve(40));
|
||||
mockOpLogStore.getUnsynced.and.returnValue(
|
||||
Promise.resolve([createMockEntry(1, 'op-1', 'client-1')]),
|
||||
);
|
||||
mockApiProvider.uploadOps.and.returnValue(
|
||||
Promise.resolve({
|
||||
results: [{ opId: 'op-1', accepted: true }],
|
||||
latestSeq: 42,
|
||||
newOps: [],
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await service.uploadPendingOps(mockApiProvider);
|
||||
|
||||
expect(mockApiProvider.setLastServerSeq).toHaveBeenCalledWith(42);
|
||||
expect(result.lastServerSeqToPersist).toBeUndefined();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -88,6 +88,10 @@ export class OperationLogUploadService {
|
|||
// We track this BEFORE decryption to detect the server's actual encryption state.
|
||||
let sawAnyPiggybackOps = false;
|
||||
let sawEncryptedPiggybackOp = false;
|
||||
// #8304: When this upload collects piggybacked ops for the caller to apply, the
|
||||
// last-server-seq covering them must be persisted by the caller AFTER those ops are
|
||||
// applied — not here. We surface the planned value instead of persisting it.
|
||||
let lastServerSeqToPersist: number | undefined;
|
||||
|
||||
await this.lockService.request(LOCK_NAMES.UPLOAD, async () => {
|
||||
// Execute pre-upload callback INSIDE the lock, BEFORE checking for pending ops.
|
||||
|
|
@ -109,6 +113,10 @@ export class OperationLogUploadService {
|
|||
let lastKnownServerSeq = await syncProvider.getLastServerSeq();
|
||||
// Track highest received sequence across ALL chunks to prevent regression
|
||||
let highestReceivedSeq = lastKnownServerSeq;
|
||||
// #8304: Becomes true once any chunk collects piggybacked ops the caller must
|
||||
// apply. From that point on we stop persisting lastServerSeq in-loop (and never
|
||||
// regress) — the caller persists the final value after processing those ops.
|
||||
let deferSeqPersistToCaller = false;
|
||||
|
||||
// Get encryption key (optional - file-based adapters handle encryption internally)
|
||||
const encryptKey = syncProvider.getEncryptKey
|
||||
|
|
@ -330,6 +338,9 @@ export class OperationLogUploadService {
|
|||
|
||||
const ops = piggybackSyncOps.map((op) => syncOpToOperation(op));
|
||||
piggybackedOps.push(...ops);
|
||||
// These ops are returned to the caller for processRemoteOps; defer the seq
|
||||
// persist so it cannot advance past ops that are not yet applied. (#8304)
|
||||
deferSeqPersistToCaller = true;
|
||||
}
|
||||
|
||||
// Update last known server seq
|
||||
|
|
@ -352,8 +363,14 @@ export class OperationLogUploadService {
|
|||
`OperationLogUploadService: hasMorePiggyback=true but no ops received, keeping lastServerSeq at ${serverSeqPlan.seqToStore}`,
|
||||
);
|
||||
}
|
||||
await syncProvider.setLastServerSeq(serverSeqPlan.seqToStore);
|
||||
lastKnownServerSeq = serverSeqPlan.seqToStore;
|
||||
if (deferSeqPersistToCaller) {
|
||||
// #8304: Hand the value to the caller; it persists after applying the
|
||||
// piggybacked ops (mirrors the download path's "persist AFTER ops stored").
|
||||
lastServerSeqToPersist = serverSeqPlan.seqToStore;
|
||||
} else {
|
||||
await syncProvider.setLastServerSeq(serverSeqPlan.seqToStore);
|
||||
}
|
||||
|
||||
// Collect rejected operations - DO NOT mark as rejected here!
|
||||
// The sync service must process piggybacked ops FIRST to allow proper conflict detection.
|
||||
|
|
@ -400,6 +417,7 @@ export class OperationLogUploadService {
|
|||
rejectedOps,
|
||||
...(hasMorePiggyback ? { hasMorePiggyback: true } : {}),
|
||||
...(piggybackHasOnlyUnencryptedData ? { piggybackHasOnlyUnencryptedData } : {}),
|
||||
...(lastServerSeqToPersist !== undefined ? { lastServerSeqToPersist } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue