From 55d3490e195bfcecbf35a952dc9e0e29ce975401 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Mon, 13 Jul 2026 21:54:07 +0200 Subject: [PATCH] fix(sync): preserve conflict cancellation and harden force overwrite (#8981) * fix(sync): honor sync import conflict outcomes Propagate nested conflict cancellation through rejected-op handling so it cannot trigger automatic merges or consume the retry budget. Report force-local resolution failures when the clean-slate upload is blocked, rejected, or accepts no operations. * fix(sync): harden force overwrite recovery Verify the exact force-upload operation, preserve clean-slate retries, and roll back rejected replacements. Keep unresolved work retryable without reporting a successful sync. * test(sync): cover clean-slate rollback in postgres --- .../src/sync/sync.service.ts | 63 +++++ ...an-slate-atomicity-sql.integration.spec.ts | 142 +++++++++++ .../tests/sync.service.spec.ts | 99 +++++++- .../imex/sync/sync-wrapper.service.spec.ts | 93 ++++++- src/app/imex/sync/sync-wrapper.service.ts | 37 ++- src/app/op-log/core/errors/sync-errors.ts | 8 + .../op-log/core/types/sync-results.types.ts | 16 +- .../sync/immediate-upload.service.spec.ts | 35 ++- .../op-log/sync/immediate-upload.service.ts | 20 +- .../sync/operation-log-sync.service.spec.ts | 150 ++++++++++- .../op-log/sync/operation-log-sync.service.ts | 20 +- ...g-upload-piggyback-seq.integration.spec.ts | 1 + .../sync/operation-log-upload.service.spec.ts | 16 ++ .../sync/operation-log-upload.service.ts | 4 +- .../sync/rejected-ops-handler.service.spec.ts | 234 ++++++++++++++++-- .../sync/rejected-ops-handler.service.ts | 86 +++++-- .../sync/server-migration.service.spec.ts | 3 +- .../op-log/sync/server-migration.service.ts | 6 +- ...ync-import-conflict-coordinator.service.ts | 56 ++++- .../ws-triggered-download.service.spec.ts | 33 ++- .../sync/ws-triggered-download.service.ts | 20 +- src/app/t.const.ts | 1 + src/assets/i18n/en.json | 1 + 23 files changed, 1054 insertions(+), 90 deletions(-) create mode 100644 packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index 81b3f4ba1d..fd1df882c6 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -60,6 +60,12 @@ const isRetryableOperationUniqueViolation = (err: unknown): boolean => { ); }; +class CleanSlateUploadRejectedError extends Error { + constructor(readonly results: UploadResult[]) { + super('Clean-slate replacement was rejected'); + } +} + /** * Main sync orchestration service. * @@ -134,6 +140,10 @@ export class SyncService { isCleanSlate?: boolean, requestStartOccupiedIds?: ReadonlySet, ): Promise { + if (isCleanSlate && ops.length === 0) { + return []; + } + const results: UploadResult[] = []; const now = Date.now(); const txStartedAt = Date.now(); @@ -146,6 +156,33 @@ export class SyncService { } } + if (isCleanSlate) { + const validations = ops.map((op) => { + const validation = + prevalidatedResults.get(op) ?? this.validationService.validateOp(op, clientId); + prevalidatedResults.set(op, validation); + return validation; + }); + if (validations.some(({ valid }) => !valid)) { + return ops.map((op, index) => { + const validation = validations[index]; + return validation.valid + ? { + opId: op.id, + accepted: false, + error: 'Clean-slate batch contains an invalid operation', + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + } + : { + opId: op.id, + accepted: false, + error: validation.error ?? 'Invalid operation', + errorCode: validation.errorCode, + }; + }); + } + } + try { // Use transaction to acquire write lock and ensure atomicity await prisma.$transaction( @@ -266,6 +303,25 @@ export class SyncService { ); } + if ( + isCleanSlate && + (results.length !== ops.length || results.some(({ accepted }) => !accepted)) + ) { + throw new CleanSlateUploadRejectedError( + ops.map((op, index) => { + const result = results[index]; + return result && !result.accepted + ? result + : { + opId: op.id, + accepted: false, + error: 'Clean-slate replacement was rolled back', + errorCode: SYNC_ERROR_CODES.INTERNAL_ERROR, + }; + }), + ); + } + // Update device last seen await tx.syncDevice.upsert({ where: { @@ -349,6 +405,13 @@ export class SyncService { batchUpload: this.config.batchUpload, }); } catch (err) { + if (err instanceof CleanSlateUploadRejectedError) { + Logger.warn( + `[user:${userId}] Clean-slate replacement rejected; existing data preserved`, + ); + return err.results; + } + // Transaction failed - all operations were rolled back const errorMessage = (err as Error).message || 'Unknown error'; diff --git a/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts b/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts new file mode 100644 index 0000000000..12d7c087c9 --- /dev/null +++ b/packages/super-sync-server/tests/integration/clean-slate-atomicity-sql.integration.spec.ts @@ -0,0 +1,142 @@ +/** + * Real-PostgreSQL coverage for destructive clean-slate uploads. + * + * Unit tests use a transaction-aware Prisma mock. This suite verifies that the + * actual database transaction restores operations, sequence state, devices, and + * storage accounting when any replacement operation is rejected. + * + * Run with: + * DATABASE_URL=postgresql://supersync:superpassword@localhost:55432/supersync_db \ + * npx vitest run --config vitest.integration.config.ts \ + * tests/integration/clean-slate-atomicity-sql.integration.spec.ts + */ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { prisma } from '../../src/db'; +import { SyncService } from '../../src/sync/sync.service'; +import { Operation, SYNC_ERROR_CODES, type SyncConfig } from '../../src/sync/sync.types'; + +const DATABASE_URL = process.env.DATABASE_URL; +const describeWithDb = DATABASE_URL ? describe : describe.skip; + +describeWithDb('Clean-slate upload atomicity (PostgreSQL)', () => { + const TEST_USER_ID = 99997; + const TEST_EMAIL = `test-clean-slate-${Date.now()}@test.local`; + const CLIENT_ID = 'clean-slate-integration-client'; + + const makeOp = (overrides: Partial = {}): Operation => ({ + id: `clean-slate-op-${Date.now()}`, + clientId: CLIENT_ID, + actionType: '[Task] Add', + opType: 'CRT', + entityType: 'TASK', + entityId: 'task-before-clean-slate', + payload: { title: 'Preserve me' }, + vectorClock: { [CLIENT_ID]: 1 }, + timestamp: Date.now(), + schemaVersion: 1, + ...overrides, + }); + + const readPersistentState = async () => { + const [operations, syncState, devices, user] = await Promise.all([ + prisma.operation.findMany({ + where: { userId: TEST_USER_ID }, + orderBy: { serverSeq: 'asc' }, + }), + prisma.userSyncState.findUniqueOrThrow({ where: { userId: TEST_USER_ID } }), + prisma.syncDevice.findMany({ + where: { userId: TEST_USER_ID }, + orderBy: { clientId: 'asc' }, + }), + prisma.user.findUniqueOrThrow({ + where: { id: TEST_USER_ID }, + select: { storageUsedBytes: true }, + }), + ]); + + return { + operations, + syncState, + devices, + storageUsedBytes: user.storageUsedBytes, + }; + }; + + beforeAll(async () => { + await prisma.user.deleteMany({ where: { id: TEST_USER_ID } }); + await prisma.user.create({ + data: { id: TEST_USER_ID, email: TEST_EMAIL, isVerified: 1 }, + }); + await prisma.userSyncState.create({ + data: { userId: TEST_USER_ID, lastSeq: 0 }, + }); + }); + + afterAll(async () => { + await prisma.user.deleteMany({ where: { id: TEST_USER_ID } }); + await prisma.$disconnect(); + }); + + beforeEach(async () => { + await prisma.operation.deleteMany({ where: { userId: TEST_USER_ID } }); + await prisma.syncDevice.deleteMany({ where: { userId: TEST_USER_ID } }); + await prisma.userSyncState.update({ + where: { userId: TEST_USER_ID }, + data: { + lastSeq: 0, + lastSnapshotSeq: null, + snapshotData: null, + snapshotAt: null, + latestFullStateSeq: null, + latestFullStateVectorClock: null, + }, + }); + await prisma.user.update({ + where: { id: TEST_USER_ID }, + data: { storageUsedBytes: 0 }, + }); + }); + + it.each([ + { name: 'serial upload path', batchUpload: false }, + { name: 'batch upload path', batchUpload: true }, + ] as const)( + 'rolls back the whole replacement on a rejected sibling ($name)', + async ({ batchUpload }) => { + const config: Partial = { batchUpload }; + const service = new SyncService(config); + const existingOp = makeOp({ id: `existing-${batchUpload}` }); + const seedResult = await service.uploadOps(TEST_USER_ID, CLIENT_ID, [existingOp]); + expect(seedResult[0].accepted).toBe(true); + + const before = await readPersistentState(); + + const replacement = makeOp({ + id: `duplicate-replacement-${batchUpload}`, + opType: 'SYNC_IMPORT', + actionType: 'LOAD_ALL_DATA', + entityType: 'ALL', + entityId: undefined, + payload: { task: { ids: ['replacement-task'] } }, + vectorClock: { [CLIENT_ID]: 2 }, + syncImportReason: 'FORCE_UPLOAD', + }); + const results = await service.uploadOps( + TEST_USER_ID, + CLIENT_ID, + [replacement, { ...replacement }], + true, + ); + + expect(results).toHaveLength(2); + expect(results.every(({ accepted }) => !accepted)).toBe(true); + expect(results[0].errorCode).toBe(SYNC_ERROR_CODES.INTERNAL_ERROR); + expect(results[1].errorCode).toBe(SYNC_ERROR_CODES.DUPLICATE_OPERATION); + + expect(await readPersistentState()).toEqual(before); + expect( + await prisma.operation.findUnique({ where: { id: replacement.id } }), + ).toBeNull(); + }, + ); +}); diff --git a/packages/super-sync-server/tests/sync.service.spec.ts b/packages/super-sync-server/tests/sync.service.spec.ts index 916b7e8926..72215bb9be 100644 --- a/packages/super-sync-server/tests/sync.service.spec.ts +++ b/packages/super-sync-server/tests/sync.service.spec.ts @@ -413,9 +413,31 @@ vi.mock('../src/db', async () => { return { prisma: { - $transaction: vi - .fn() - .mockImplementation(async (callback: any) => callback(createTxMock())), + $transaction: vi.fn().mockImplementation(async (callback: any) => { + const transactionStart = { + operations: new Map( + Array.from(state.operations, ([id, op]) => [id, { ...op }]), + ), + syncDevices: new Map( + Array.from(state.syncDevices, ([id, device]) => [id, { ...device }]), + ), + userSyncStates: new Map( + Array.from(state.userSyncStates, ([id, syncState]) => [id, { ...syncState }]), + ), + users: new Map(Array.from(state.users, ([id, user]) => [id, { ...user }])), + serverSeqCounter: state.serverSeqCounter, + }; + try { + return await callback(createTxMock()); + } catch (error) { + state.operations = transactionStart.operations; + state.syncDevices = transactionStart.syncDevices; + state.userSyncStates = transactionStart.userSyncStates; + state.users = transactionStart.users; + state.serverSeqCounter = transactionStart.serverSeqCounter; + throw error; + } + }), operation: { findFirst: vi.fn().mockImplementation(async (args: any) => { if (args.where?.opType?.in) { @@ -755,6 +777,77 @@ describe('SyncService', () => { expect(latestSeq).toBe(1); }); + it('preserves existing data when a clean-slate replacement fails validation', async () => { + const service = new SyncService({ maxPayloadSizeBytes: 500 }); + const existingOp = makeOp({ + id: 'existing-before-clean-slate', + payload: { title: 'Keep me' }, + }); + const invalidReplacement = makeOp({ + id: 'invalid-clean-slate-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + payload: { data: 'x'.repeat(1_000) }, + }); + + const initialResult = await service.uploadOps(userId, clientId, [existingOp]); + vi.mocked(prisma.$transaction).mockClear(); + const replacementResult = await service.uploadOps( + userId, + clientId, + [invalidReplacement], + true, + ); + + expect(initialResult[0].accepted).toBe(true); + expect(replacementResult[0]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.PAYLOAD_TOO_LARGE, + }), + ); + expect(testState.operations.has(existingOp.id)).toBe(true); + expect(testState.operations.has(invalidReplacement.id)).toBe(false); + expect(prisma.$transaction).not.toHaveBeenCalled(); + }); + + it('preserves existing data when any clean-slate operation is rejected', async () => { + const service = getSyncService(); + const existingOp = makeOp({ id: 'existing-before-rejected-clean-slate' }); + const replacement = makeOp({ + id: 'duplicate-clean-slate-replacement', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: undefined, + }); + await service.uploadOps(userId, clientId, [existingOp]); + + const results = await service.uploadOps( + userId, + clientId, + [replacement, { ...replacement }], + true, + ); + + expect(results).toHaveLength(2); + expect(results.every(({ accepted }) => !accepted)).toBe(true); + expect(results[1].errorCode).toBe(SYNC_ERROR_CODES.DUPLICATE_OPERATION); + expect(testState.operations.has(existingOp.id)).toBe(true); + expect(testState.operations.has(replacement.id)).toBe(false); + }); + + it('does not wipe existing data for an empty clean-slate upload', async () => { + const service = getSyncService(); + const existingOp = makeOp({ id: 'existing-before-empty-clean-slate' }); + await service.uploadOps(userId, clientId, [existingOp]); + + const results = await service.uploadOps(userId, clientId, [], true); + + expect(results).toEqual([]); + expect(testState.operations.has(existingOp.id)).toBe(true); + }); + it('should handle multiple operations in order', async () => { const service = getSyncService(); const ops: Operation[] = [ diff --git a/src/app/imex/sync/sync-wrapper.service.spec.ts b/src/app/imex/sync/sync-wrapper.service.spec.ts index f49dae4ed2..f40b6736cf 100644 --- a/src/app/imex/sync/sync-wrapper.service.spec.ts +++ b/src/app/imex/sync/sync-wrapper.service.spec.ts @@ -38,6 +38,8 @@ import { UploadRevToMatchMismatchAPIError, WebDavNativeRequestError, EncryptNoPasswordError, + ForceUploadFailedError, + ForceUploadPendingOpsError, IncompleteRemoteOperationsError, } from '../../op-log/core/errors/sync-errors'; import { DialogEnterEncryptionPasswordComponent } from './dialog-enter-encryption-password/dialog-enter-encryption-password.component'; @@ -1600,7 +1602,7 @@ describe('SyncWrapperService', () => { mockSyncService.forceUploadLocalState = jasmine .createSpy('forceUploadLocalState') - .and.resolveTo(); + .and.resolveTo({ hasUnresolvedOps: false }); await service.sync(); @@ -1622,7 +1624,7 @@ describe('SyncWrapperService', () => { mockSyncService.forceUploadLocalState = jasmine .createSpy('forceUploadLocalState') - .and.resolveTo(); + .and.resolveTo({ hasUnresolvedOps: false }); const result = await service.sync(); @@ -1632,6 +1634,29 @@ describe('SyncWrapperService', () => { expect(result).toBe(SyncStatus.InSync); }); + it('should leave sync pending when the overwrite succeeds with unresolved later ops', async () => { + const conflictError = new LocalDataConflictError( + 2, + { tasks: [] }, + { clientB: 3 }, + ); + mockSyncService.downloadRemoteOps.and.rejectWith(conflictError); + mockMatDialog.open.and.returnValue({ + afterClosed: () => of('USE_LOCAL'), + } as any); + mockSyncService.forceUploadLocalState = jasmine + .createSpy('forceUploadLocalState') + .and.resolveTo({ hasUnresolvedOps: true }); + + const result = await service.sync(); + + expect(result).toBe('HANDLED_ERROR'); + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith( + 'UNKNOWN_OR_CHANGED', + ); + expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC'); + }); + it('should call forceDownloadRemoteState when user chooses USE_REMOTE', async () => { const conflictError = new LocalDataConflictError( 2, @@ -1701,11 +1726,37 @@ describe('SyncWrapperService', () => { const result = await service.sync(); expect(result).toBe('HANDLED_ERROR'); + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); expect(mockSnackService.open).toHaveBeenCalledWith( - jasmine.objectContaining({ - type: 'ERROR', - }), + jasmine.objectContaining({ type: 'ERROR' }), ); + expect(mockSnackService.open).not.toHaveBeenCalledWith( + jasmine.objectContaining({ msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED }), + ); + }); + + it('should translate a typed force-upload failure during conflict resolution', async () => { + const conflictError = new LocalDataConflictError( + 2, + { tasks: [] }, + { clientB: 3 }, + ); + mockSyncService.downloadRemoteOps.and.rejectWith(conflictError); + mockMatDialog.open.and.returnValue({ + afterClosed: () => of('USE_LOCAL'), + } as any); + mockSyncService.forceUploadLocalState = jasmine + .createSpy('forceUploadLocalState') + .and.rejectWith(new ForceUploadFailedError()); + + const result = await service.sync(); + + expect(result).toBe('HANDLED_ERROR'); + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); + expect(mockSnackService.open).toHaveBeenCalledWith({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); }); // GHSA-9544-hjjr-fg8h: USE_LOCAL force-uploads, which refuses to send @@ -1790,6 +1841,9 @@ describe('SyncWrapperService', () => { type: 'ERROR', }), ); + expect(mockSnackService.open).not.toHaveBeenCalledWith( + jasmine.objectContaining({ msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED }), + ); }); it('should return HANDLED_ERROR when provider becomes unavailable during resolution', async () => { @@ -1836,7 +1890,7 @@ describe('SyncWrapperService', () => { mockSyncService.forceUploadLocalState = jasmine .createSpy('forceUploadLocalState') - .and.resolveTo(); + .and.resolveTo({ hasUnresolvedOps: false }); await service.sync(); @@ -1902,6 +1956,33 @@ describe('SyncWrapperService', () => { ); }); + it('should translate FORCE_UPLOAD failures raised during op-log sync', async () => { + mockSyncService.uploadPendingOps.and.rejectWith(new ForceUploadFailedError()); + + const result = await service.sync(); + + expect(result).toBe('HANDLED_ERROR'); + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); + expect(mockSnackService.open).toHaveBeenCalledWith({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + }); + + it('should keep sync pending when nested force upload has unresolved ops', async () => { + mockSyncService.uploadPendingOps.and.rejectWith(new ForceUploadPendingOpsError()); + + const result = await service.sync(); + + expect(result).toBe('HANDLED_ERROR'); + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith( + 'UNKNOWN_OR_CHANGED', + ); + expect(mockSnackService.open).not.toHaveBeenCalledWith( + jasmine.objectContaining({ type: 'ERROR' }), + ); + }); + it('should preserve a persistent recovery action when sync rethrows', async () => { mockSnackService.hasPendingPersistentAction.and.returnValue(true); mockSyncService.downloadRemoteOps.and.rejectWith( diff --git a/src/app/imex/sync/sync-wrapper.service.ts b/src/app/imex/sync/sync-wrapper.service.ts index cebb6db894..6494407ece 100644 --- a/src/app/imex/sync/sync-wrapper.service.ts +++ b/src/app/imex/sync/sync-wrapper.service.ts @@ -24,6 +24,8 @@ import { IncompleteRemoteOperationsError, SyncDataCorruptedError, UploadRevToMatchMismatchAPIError, + ForceUploadFailedError, + ForceUploadPendingOpsError, } from '../../op-log/core/errors/sync-errors'; import { MAX_LWW_REUPLOAD_RETRIES } from '../../op-log/core/operation-log.const'; import { SyncConfig } from '../../features/config/global-config.model'; @@ -960,6 +962,16 @@ export class SyncWrapperService { config: { duration: 15000 }, }); return 'HANDLED_ERROR'; + } else if (error instanceof ForceUploadPendingOpsError) { + this._providerManager.setSyncStatus('UNKNOWN_OR_CHANGED'); + return 'HANDLED_ERROR'; + } else if (error instanceof ForceUploadFailedError) { + this._providerManager.setSyncStatus('ERROR'); + this._snackService.open({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + return 'HANDLED_ERROR'; } else { this._providerManager.setSyncStatus('ERROR'); const errStr = getSyncErrorStr(error); @@ -1052,8 +1064,11 @@ export class SyncWrapperService { return; } - await this._opLogSyncService.forceUploadLocalState(syncCapableProvider); - this._providerManager.setSyncStatus('IN_SYNC'); + const forceUploadResult = + await this._opLogSyncService.forceUploadLocalState(syncCapableProvider); + this._providerManager.setSyncStatus( + forceUploadResult.hasUnresolvedOps ? 'UNKNOWN_OR_CHANGED' : 'IN_SYNC', + ); SyncLog.log('SyncWrapperService: Force upload complete'); } catch (error) { // GHSA-9544-hjjr-fg8h: a keyless-but-encryption-enabled provider makes @@ -1064,9 +1079,9 @@ export class SyncWrapperService { this._handleMissingPasswordDialog(); } else { SyncLog.err('SyncWrapperService: Force upload failed:', error); - const errStr = getSyncErrorStr(error); + this._providerManager.setSyncStatus('ERROR'); this._snackService.open({ - msg: errStr, + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, type: 'ERROR', }); } @@ -1330,7 +1345,12 @@ export class SyncWrapperService { SyncLog.log( 'SyncWrapperService: User chose USE_LOCAL - uploading local state to overwrite remote', ); - await this._opLogSyncService.forceUploadLocalState(syncCapableProvider); + const forceUploadResult = + await this._opLogSyncService.forceUploadLocalState(syncCapableProvider); + if (forceUploadResult.hasUnresolvedOps) { + this._providerManager.setSyncStatus('UNKNOWN_OR_CHANGED'); + return 'HANDLED_ERROR'; + } this._providerManager.setSyncStatus('IN_SYNC'); return SyncStatus.InSync; } else if (resolution === 'USE_REMOTE') { @@ -1371,9 +1391,12 @@ export class SyncWrapperService { 'SyncWrapperService: Error during conflict resolution:', resolutionError, ); - const errStr = getSyncErrorStr(resolutionError); + this._providerManager.setSyncStatus('ERROR'); this._snackService.open({ - msg: errStr, + msg: + resolutionError instanceof ForceUploadFailedError + ? T.F.SYNC.S.FORCE_UPLOAD_FAILED + : getSyncErrorStr(resolutionError), type: 'ERROR', }); return 'HANDLED_ERROR'; diff --git a/src/app/op-log/core/errors/sync-errors.ts b/src/app/op-log/core/errors/sync-errors.ts index f0fefbf3bf..9d6f347646 100644 --- a/src/app/op-log/core/errors/sync-errors.ts +++ b/src/app/op-log/core/errors/sync-errors.ts @@ -112,6 +112,14 @@ export class UnknownSyncStateError extends Error { override name = 'UnknownSyncStateError'; } +export class ForceUploadFailedError extends Error { + override name = 'ForceUploadFailedError'; +} + +export class ForceUploadPendingOpsError extends Error { + override name = 'ForceUploadPendingOpsError'; +} + /** * A deferred action can never be persisted (invalid entity identifiers or an * invalid operation payload) — a deterministic condition, not a transient diff --git a/src/app/op-log/core/types/sync-results.types.ts b/src/app/op-log/core/types/sync-results.types.ts index 1ef8e4ea7b..0796239850 100644 --- a/src/app/op-log/core/types/sync-results.types.ts +++ b/src/app/op-log/core/types/sync-results.types.ts @@ -214,11 +214,17 @@ export interface UploadOptions { * SyncSessionValidationService latch — the wrapper reads it once before * deciding IN_SYNC vs ERROR. (#7330) */ -export interface DownloadResultForRejection { - newOpsCount: number; - allOpClocks?: VectorClock[]; - snapshotVectorClock?: VectorClock; -} +export type DownloadResultForRejection = + | { + kind: 'completed'; + newOpsCount: number; + allOpClocks?: VectorClock[]; + snapshotVectorClock?: VectorClock; + } + | { + /** User declined the nested SYNC_IMPORT conflict resolution. */ + kind: 'cancelled'; + }; /** * Callback type for triggering downloads during concurrent modification resolution. diff --git a/src/app/op-log/sync/immediate-upload.service.spec.ts b/src/app/op-log/sync/immediate-upload.service.spec.ts index 9c08ea3e04..1e7fe02cab 100644 --- a/src/app/op-log/sync/immediate-upload.service.spec.ts +++ b/src/app/op-log/sync/immediate-upload.service.spec.ts @@ -10,7 +10,11 @@ import { SyncCycleGuardService } from './sync-cycle-guard.service'; import { BehaviorSubject } from 'rxjs'; import { RejectedOpInfo } from '../core/types/sync-results.types'; import { SnackService } from '../../core/snack/snack.service'; -import { IncompleteRemoteOperationsError } from '../core/errors/sync-errors'; +import { + ForceUploadFailedError, + ForceUploadPendingOpsError, + IncompleteRemoteOperationsError, +} from '../core/errors/sync-errors'; import { T } from '../../t.const'; describe('ImmediateUploadService', () => { @@ -200,6 +204,35 @@ describe('ImmediateUploadService', () => { expect(mockSnackService.open).not.toHaveBeenCalled(); })); + it('should surface a force-upload failure raised by conflict resolution', fakeAsync(() => { + mockSyncService.uploadPendingOps.and.rejectWith(new ForceUploadFailedError()); + + service.initialize(); + service.trigger(); + tick(2000); + flush(); + + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); + expect(mockSnackService.open).toHaveBeenCalledWith({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + })); + + it('should keep sync pending when force upload leaves unresolved ops', fakeAsync(() => { + mockSyncService.uploadPendingOps.and.rejectWith(new ForceUploadPendingOpsError()); + + service.initialize(); + service.trigger(); + tick(2000); + flush(); + + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith( + 'UNKNOWN_OR_CHANGED', + ); + expect(mockSnackService.open).not.toHaveBeenCalled(); + })); + it('should report UNKNOWN_OR_CHANGED when a local-win follow-up lacks a mandatory encryption key', fakeAsync(() => { mockSyncService.uploadPendingOps.and.returnValues( Promise.resolve(completedResult({ uploadedCount: 1, localWinOpsCreated: 1 })), diff --git a/src/app/op-log/sync/immediate-upload.service.ts b/src/app/op-log/sync/immediate-upload.service.ts index d2440aa2da..e0d803b8d6 100644 --- a/src/app/op-log/sync/immediate-upload.service.ts +++ b/src/app/op-log/sync/immediate-upload.service.ts @@ -11,7 +11,11 @@ import { handleStorageQuotaError } from './sync-error-utils'; import { SyncWrapperService } from '../../imex/sync/sync-wrapper.service'; import { SyncSessionValidationService } from './sync-session-validation.service'; import { SyncCycleGuardService } from './sync-cycle-guard.service'; -import { IncompleteRemoteOperationsError } from '../core/errors/sync-errors'; +import { + ForceUploadFailedError, + ForceUploadPendingOpsError, + IncompleteRemoteOperationsError, +} from '../core/errors/sync-errors'; import { SnackService } from '../../core/snack/snack.service'; import { T } from '../../t.const'; @@ -328,6 +332,20 @@ export class ImmediateUploadService implements OnDestroy { ); } } catch (e) { + if (e instanceof ForceUploadPendingOpsError) { + this._providerManager.setSyncStatus('UNKNOWN_OR_CHANGED'); + return; + } + + if (e instanceof ForceUploadFailedError) { + this._providerManager.setSyncStatus('ERROR'); + this._snackService.open({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + return; + } + if (e instanceof IncompleteRemoteOperationsError) { this._providerManager.setSyncStatus('ERROR'); if (!this._snackService.hasPendingPersistentAction()) { diff --git a/src/app/op-log/sync/operation-log-sync.service.spec.ts b/src/app/op-log/sync/operation-log-sync.service.spec.ts index 6235668fa4..697cfc9cb0 100644 --- a/src/app/op-log/sync/operation-log-sync.service.spec.ts +++ b/src/app/op-log/sync/operation-log-sync.service.spec.ts @@ -36,6 +36,9 @@ import { } from '../core/operation.types'; import { TranslateService } from '@ngx-translate/core'; import { + EncryptNoPasswordError, + ForceUploadFailedError, + ForceUploadPendingOpsError, IncompleteRemoteOperationsError, LocalDataConflictError, } from '../core/errors/sync-errors'; @@ -207,6 +210,7 @@ describe('OperationLogSyncService', () => { 'handleRejectedOps', ]); rejectedOpsHandlerServiceSpy.handleRejectedOps.and.resolveTo({ + kind: 'completed', mergedOpsCreated: 0, permanentRejectionCount: 0, }); @@ -802,7 +806,11 @@ describe('OperationLogSyncService', () => { rejectedOpsHandlerServiceSpy.handleRejectedOps.and.callFake( async (_ops, callback) => { capturedCallback = callback; - return { mergedOpsCreated: 0, permanentRejectionCount: 0 }; + return { + kind: 'completed', + mergedOpsCreated: 0, + permanentRejectionCount: 0, + }; }, ); @@ -831,6 +839,39 @@ describe('OperationLogSyncService', () => { }); }); + it('should propagate nested download cancellation as a cancelled upload', async () => { + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 0, + piggybackedOps: [], + rejectedCount: 1, + rejectedOps: [ + { + opId: 'local-op-1', + error: 'Concurrent', + errorCode: 'CONFLICT_CONCURRENT', + }, + ], + }); + spyOn(service, 'downloadRemoteOps').and.resolveTo({ kind: 'cancelled' }); + rejectedOpsHandlerServiceSpy.handleRejectedOps.and.callFake( + async (_ops, callback) => { + const nestedResult = await callback?.(); + if (nestedResult?.kind === 'cancelled') { + return { kind: 'cancelled' }; + } + return { + kind: 'completed', + mergedOpsCreated: 0, + permanentRejectionCount: 0, + }; + }, + ); + + const result = await service.uploadPendingOps(mockProvider); + + expect(result.kind).toBe('cancelled'); + }); + it('should add mergedOpsFromRejection to localWinOpsCreated in result', async () => { const piggybackedOp: Operation = { id: 'piggybacked-1', @@ -871,6 +912,7 @@ describe('OperationLogSyncService', () => { // handleRejectedOps returns 3 merged ops created rejectedOpsHandlerServiceSpy.handleRejectedOps.and.resolveTo({ + kind: 'completed', mergedOpsCreated: 3, permanentRejectionCount: 0, }); @@ -1007,7 +1049,11 @@ describe('OperationLogSyncService', () => { // resolution. The latch is flipped inside the nested download's // validateAfterSync — here we just exercise the call. await callback?.(); - return { mergedOpsCreated: 0, permanentRejectionCount: 0 }; + return { + kind: 'completed', + mergedOpsCreated: 0, + permanentRejectionCount: 0, + }; }, ); @@ -2810,6 +2856,10 @@ describe('OperationLogSyncService', () => { rejectedOps: [], localWinOpsCreated: 0, }); + serverMigrationServiceSpy.handleServerMigration.and.resolveTo('force-import'); + opLogStoreSpy.getOpById.and.resolveTo({ + syncedAt: Date.now(), + } as OperationLogEntry); }); it('should call handleServerMigration to create SYNC_IMPORT', async () => { @@ -2818,18 +2868,21 @@ describe('OperationLogSyncService', () => { setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), } as unknown as OperationSyncCapable; - await service.forceUploadLocalState(mockProvider); + const result = await service.forceUploadLocalState(mockProvider); expect(serverMigrationServiceSpy.handleServerMigration).toHaveBeenCalledWith( mockProvider, { skipServerEmptyCheck: true, syncImportReason: 'FORCE_UPLOAD' }, ); + expect(opLogStoreSpy.getOpById).toHaveBeenCalledOnceWith('force-import'); + expect(result).toEqual({ hasUnresolvedOps: false }); }); it('should upload pending ops after creating SYNC_IMPORT', async () => { const callOrder: string[] = []; serverMigrationServiceSpy.handleServerMigration.and.callFake(async () => { callOrder.push('handleServerMigration'); + return 'force-import'; }); uploadServiceSpy.uploadPendingOps.and.callFake(async () => { callOrder.push('uploadPendingOps'); @@ -2865,6 +2918,19 @@ describe('OperationLogSyncService', () => { ); }); + it('should reject when no FORCE_UPLOAD operation was created', async () => { + serverMigrationServiceSpy.handleServerMigration.and.resolveTo(); + + const mockProvider = { + supportsOperationSync: true, + } as unknown as OperationSyncCapable; + + await expectAsync( + service.forceUploadLocalState(mockProvider), + ).toBeRejectedWithError(ForceUploadFailedError); + expect(uploadServiceSpy.uploadPendingOps).not.toHaveBeenCalled(); + }); + it('should propagate errors from uploadPendingOps', async () => { const error = new Error('Upload failed'); uploadServiceSpy.uploadPendingOps.and.rejectWith(error); @@ -2879,6 +2945,75 @@ describe('OperationLogSyncService', () => { ); }); + it('should reject when mandatory encryption blocks the force upload', async () => { + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 0, + piggybackedOps: [], + rejectedCount: 0, + rejectedOps: [], + encryptionRequiredKeyMissing: true, + }); + + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await expectAsync( + service.forceUploadLocalState(mockProvider), + ).toBeRejectedWithError(EncryptNoPasswordError); + }); + + it('should reject with a typed error when the FORCE_UPLOAD op is not acknowledged', async () => { + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 0, + piggybackedOps: [], + rejectedCount: 1, + rejectedOps: [ + { + opId: 'force-import', + error: 'snapshot rejected', + errorCode: 'VALIDATION_ERROR', + }, + ], + }); + opLogStoreSpy.getOpById.and.resolveTo({} as OperationLogEntry); + + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await expectAsync( + service.forceUploadLocalState(mockProvider), + ).toBeRejectedWithError(ForceUploadFailedError); + }); + + it('should succeed when the FORCE_UPLOAD op is accepted despite an unrelated rejection', async () => { + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 1, + piggybackedOps: [], + rejectedCount: 1, + rejectedOps: [ + { + opId: 'older-op', + error: 'superseded', + errorCode: 'VALIDATION_ERROR', + }, + ], + localWinOpsCreated: 0, + }); + + const mockProvider = { + supportsOperationSync: true, + } as unknown as OperationSyncCapable; + + const result = await service.forceUploadLocalState(mockProvider); + + expect(result).toEqual({ hasUnresolvedOps: true }); + expect(opLogStoreSpy.getOpById).toHaveBeenCalledOnceWith('force-import'); + }); + it('should upload with isCleanSlate=true to delete server data before accepting new data', async () => { // This is critical for recovery scenarios like decrypt errors where the server // may have data encrypted with a different password. Clean slate ensures the @@ -5337,7 +5472,9 @@ describe('OperationLogSyncService', () => { syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('USE_LOCAL'); - const forceUploadSpy = spyOn(service, 'forceUploadLocalState').and.resolveTo(); + const forceUploadSpy = spyOn(service, 'forceUploadLocalState').and.resolveTo({ + hasUnresolvedOps: false, + }); const mockProvider = { isReady: () => Promise.resolve(true), @@ -5346,6 +5483,11 @@ describe('OperationLogSyncService', () => { await service.uploadPendingOps(mockProvider); expect(forceUploadSpy).toHaveBeenCalledWith(mockProvider); + + forceUploadSpy.and.resolveTo({ hasUnresolvedOps: true }); + await expectAsync(service.uploadPendingOps(mockProvider)).toBeRejectedWithError( + ForceUploadPendingOpsError, + ); }); it('should call forceDownloadRemoteState when user chooses USE_REMOTE', async () => { diff --git a/src/app/op-log/sync/operation-log-sync.service.ts b/src/app/op-log/sync/operation-log-sync.service.ts index 73e639ea07..4e030aae0c 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -60,7 +60,10 @@ import { SyncProviderManager } from '../sync-providers/provider-manager.service' import { getDefaultMainModelData, MODEL_CONFIGS } from '../model/model-config'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; import { SyncLocalStateService } from './sync-local-state.service'; -import { SyncImportConflictCoordinatorService } from './sync-import-conflict-coordinator.service'; +import { + ForceUploadResult, + SyncImportConflictCoordinatorService, +} from './sync-import-conflict-coordinator.service'; import { isExampleTaskCreateOp } from '../validation/is-example-task-op.util'; import { Operation, OperationLogEntry } from '../core/operation.types'; import { ValidateStateService } from '../validation/validate-state.service'; @@ -289,6 +292,7 @@ export class OperationLogSyncService { // state and get re-uploaded infinitely. let localWinOpsCreated = 0; let rejectionResult: RejectionHandlingResult = { + kind: 'completed', mergedOpsCreated: 0, permanentRejectionCount: 0, }; @@ -439,6 +443,7 @@ export class OperationLogSyncService { switch (outcome.kind) { case 'ops_processed': return { + kind: 'completed', newOpsCount: outcome.newOpsCount, allOpClocks: outcome.allOpClocks, snapshotVectorClock: outcome.snapshotVectorClock, @@ -446,13 +451,15 @@ export class OperationLogSyncService { case 'no_new_ops': case 'snapshot_hydrated': return { + kind: 'completed', newOpsCount: 0, allOpClocks: outcome.allOpClocks, snapshotVectorClock: outcome.snapshotVectorClock, }; case 'server_migration_handled': + return { kind: 'completed', newOpsCount: 0 }; case 'cancelled': - return { newOpsCount: 0 }; + return { kind: 'cancelled' }; case 'blocked_incompatible': throw new Error('Nested download blocked by an incompatible remote operation.'); } @@ -462,6 +469,9 @@ export class OperationLogSyncService { result.rejectedOps, downloadCallback, ); + if (rejectionResult.kind === 'cancelled') { + return { kind: 'cancelled' }; + } localWinOpsCreated += rejectionResult.mergedOpsCreated; } catch (rejectionError) { // FIX #6571: Propagate rejection handler errors instead of swallowing them. @@ -1380,8 +1390,10 @@ export class OperationLogSyncService { * * @param syncProvider - The sync provider to upload to */ - async forceUploadLocalState(syncProvider: OperationSyncCapable): Promise { - await this.syncImportConflictCoordinator.forceUploadLocalState(syncProvider); + async forceUploadLocalState( + syncProvider: OperationSyncCapable, + ): Promise { + return this.syncImportConflictCoordinator.forceUploadLocalState(syncProvider); } /** diff --git a/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts b/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts index 65c973f4f8..db489ed5dd 100644 --- a/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts +++ b/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts @@ -234,6 +234,7 @@ describe('OperationLogSyncService + OperationLogUploadService — piggyback seq ['handleRejectedOps'], ); rejectedOpsHandlerServiceSpy.handleRejectedOps.and.resolveTo({ + kind: 'completed', mergedOpsCreated: 0, permanentRejectionCount: 0, }); diff --git a/src/app/op-log/sync/operation-log-upload.service.spec.ts b/src/app/op-log/sync/operation-log-upload.service.spec.ts index c7bb616106..0123d76f14 100644 --- a/src/app/op-log/sync/operation-log-upload.service.spec.ts +++ b/src/app/op-log/sync/operation-log-upload.service.spec.ts @@ -1199,6 +1199,22 @@ describe('OperationLogUploadService', () => { expect(callArgs[7]).toBeUndefined(); }); + it('should preserve clean-slate intent when retrying a FORCE_UPLOAD SyncImport', async () => { + const entry = createFullStateEntry( + 1, + 'force-import', + 'client-1', + OpType.SyncImport, + ); + entry.op.syncImportReason = 'FORCE_UPLOAD'; + mockOpLogStore.getUnsynced.and.returnValue(Promise.resolve([entry])); + + await service.uploadPendingOps(mockApiProvider); + + const callArgs = mockApiProvider.uploadSnapshot.calls.mostRecent().args; + expect(callArgs[7]).toBe(true); + }); + it('should still upload regular ops when full-state op is rejected', async () => { const fullStateEntry = createFullStateEntry( 1, diff --git a/src/app/op-log/sync/operation-log-upload.service.ts b/src/app/op-log/sync/operation-log-upload.service.ts index 31d71fbbb3..cd08b120b1 100644 --- a/src/app/op-log/sync/operation-log-upload.service.ts +++ b/src/app/op-log/sync/operation-log-upload.service.ts @@ -227,7 +227,9 @@ export class OperationLogUploadService { // BackupImport/Repair: always wipe server (recovery operations replace all state) // SyncImport: only wipe when explicitly requested (preserves SYNC_IMPORT_EXISTS check) const isCleanSlateForOp = - entry.op.opType === OpType.SyncImport ? options?.isCleanSlate : true; + entry.op.opType === OpType.SyncImport + ? entry.op.syncImportReason === 'FORCE_UPLOAD' || options?.isCleanSlate + : true; const result = await this._uploadFullStateOpAsSnapshot( syncProvider, entry, diff --git a/src/app/op-log/sync/rejected-ops-handler.service.spec.ts b/src/app/op-log/sync/rejected-ops-handler.service.spec.ts index c7cab790cb..3674d11415 100644 --- a/src/app/op-log/sync/rejected-ops-handler.service.spec.ts +++ b/src/app/op-log/sync/rejected-ops-handler.service.spec.ts @@ -70,7 +70,11 @@ describe('RejectedOpsHandlerService', () => { describe('handleRejectedOps', () => { it('should return zero counts when no rejected ops provided', async () => { const result = await service.handleRejectedOps([]); - expect(result).toEqual({ mergedOpsCreated: 0, permanentRejectionCount: 0 }); + expect(result).toEqual({ + kind: 'completed', + mergedOpsCreated: 0, + permanentRejectionCount: 0, + }); }); it('should skip already synced ops', async () => { @@ -286,12 +290,13 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ serverClient: 10 }], snapshotVectorClock: { serverClient: 10 }, } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -317,7 +322,7 @@ describe('RejectedOpsHandlerService', () => { const op = createOp({ id: 'superseded-op-1' }); opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); downloadCallback.and.returnValue( - Promise.resolve({ newOpsCount: 1 } as DownloadResultForRejection), + Promise.resolve({ kind: 'completed', newOpsCount: 1 }), ); await service.handleRejectedOps( @@ -342,12 +347,13 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ serverClient: 10 }], snapshotVectorClock: { serverClient: 10 }, } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -381,7 +387,7 @@ describe('RejectedOpsHandlerService', () => { const op = createOp({ id: 'op-1' }); opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); downloadCallback.and.returnValue( - Promise.resolve({ newOpsCount: 1 } as DownloadResultForRejection), + Promise.resolve({ kind: 'completed', newOpsCount: 1 }), ); await service.handleRejectedOps( @@ -392,6 +398,177 @@ describe('RejectedOpsHandlerService', () => { expect(downloadCallback).toHaveBeenCalled(); }); + it('should stop rejection handling when the nested download is cancelled', async () => { + const op = createOp({ id: 'op-1' }); + opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); + downloadCallback.and.returnValue(Promise.resolve({ kind: 'cancelled' })); + + const result = await service.handleRejectedOps( + [ + { + opId: 'op-1', + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT', + existingClock: { serverClient: 2 }, + }, + ], + downloadCallback, + ); + + expect(result).toEqual({ kind: 'cancelled' }); + expect(downloadCallback).toHaveBeenCalledTimes(1); + expect( + supersededOperationResolverSpy.resolveSupersededLocalOps, + ).not.toHaveBeenCalled(); + expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled(); + }); + + it('should stop rejection handling when the forced download is cancelled', async () => { + const op = createOp({ id: 'op-1' }); + opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); + downloadCallback.and.callFake(async (options) => { + if (options?.forceFromSeq0) { + return { kind: 'cancelled' }; + } + return { kind: 'completed', newOpsCount: 0 }; + }); + + const result = await service.handleRejectedOps( + [ + { + opId: 'op-1', + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT', + existingClock: { serverClient: 2 }, + }, + ], + downloadCallback, + ); + + expect(result).toEqual({ kind: 'cancelled' }); + expect(downloadCallback).toHaveBeenCalledTimes(2); + expect( + supersededOperationResolverSpy.resolveSupersededLocalOps, + ).not.toHaveBeenCalled(); + expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled(); + }); + + it('should not consume the resolution-attempt budget when downloads are cancelled', async () => { + const op = createOp({ id: 'op-1' }); + opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); + downloadCallback.and.returnValue(Promise.resolve({ kind: 'cancelled' })); + + for (let i = 0; i <= MAX_CONCURRENT_RESOLUTION_ATTEMPTS; i++) { + await service.handleRejectedOps( + [ + { + opId: 'op-1', + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT', + existingClock: { serverClient: 2 }, + }, + ], + downloadCallback, + ); + } + + expect(downloadCallback).toHaveBeenCalledTimes( + MAX_CONCURRENT_RESOLUTION_ATTEMPTS + 1, + ); + expect( + supersededOperationResolverSpy.resolveSupersededLocalOps, + ).not.toHaveBeenCalled(); + expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled(); + }); + + it('should not consume the resolution-attempt budget when forced downloads are cancelled', async () => { + const op = createOp({ id: 'op-1' }); + opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); + downloadCallback.and.callFake(async (options) => + options?.forceFromSeq0 + ? { kind: 'cancelled' } + : { kind: 'completed', newOpsCount: 0 }, + ); + + for (let i = 0; i <= MAX_CONCURRENT_RESOLUTION_ATTEMPTS; i++) { + await service.handleRejectedOps( + [ + { + opId: 'op-1', + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT', + existingClock: { serverClient: 2 }, + }, + ], + downloadCallback, + ); + } + + expect(downloadCallback).toHaveBeenCalledTimes( + (MAX_CONCURRENT_RESOLUTION_ATTEMPTS + 1) * 2, + ); + expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled(); + }); + + it('should keep retry-exceeded rejection terminal when another conflict is cancelled', async () => { + const entriesById = new Map>(); + opLogStoreSpy.getOpById.and.callFake(async (opId) => entriesById.get(opId)); + opLogStoreSpy.markRejected.and.resolveTo(); + supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); + downloadCallback.and.callFake(async (options) => + options?.forceFromSeq0 + ? { + kind: 'completed', + newOpsCount: 0, + allOpClocks: [{ remoteClient: 2 }], + } + : { kind: 'completed', newOpsCount: 0 }, + ); + + for (let i = 0; i < MAX_CONCURRENT_RESOLUTION_ATTEMPTS; i++) { + const op = createOp({ + id: `at-limit-${i}`, + entityId: 'at-limit-entity', + }); + entriesById.set(op.id, mockEntry(op)); + await service.handleRejectedOps( + [ + { + opId: op.id, + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT', + }, + ], + downloadCallback, + ); + } + + const atLimitOp = createOp({ + id: 'at-limit-cancelled', + entityId: 'at-limit-entity', + }); + const resolvableOp = createOp({ + id: 'resolvable-cancelled', + entityId: 'resolvable-entity', + }); + entriesById.set(atLimitOp.id, mockEntry(atLimitOp)); + entriesById.set(resolvableOp.id, mockEntry(resolvableOp)); + opLogStoreSpy.markRejected.calls.reset(); + downloadCallback.and.resolveTo({ kind: 'cancelled' }); + + const cancelledResult = await service.handleRejectedOps( + [atLimitOp, resolvableOp].map((op) => ({ + opId: op.id, + error: 'concurrent', + errorCode: 'CONFLICT_CONCURRENT' as const, + })), + downloadCallback, + ); + + expect(cancelledResult).toEqual({ kind: 'cancelled' }); + expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([atLimitOp.id]); + }); + it('should NOT reject pending ops when the download throws transiently, and re-throw (#8331)', async () => { // REGRESSION TEST: A network blip during the concurrent-modification // download must not permanently drop the user's pending local edits. @@ -448,11 +625,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); await service.handleRejectedOps( @@ -472,12 +650,13 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [remoteClock], snapshotVectorClock: { snapshot: 1 }, } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -493,7 +672,11 @@ describe('RejectedOpsHandlerService', () => { [remoteClock], { snapshot: 1 }, ); - expect(result).toEqual({ mergedOpsCreated: 1, permanentRejectionCount: 0 }); + expect(result).toEqual({ + kind: 'completed', + mergedOpsCreated: 1, + permanentRejectionCount: 0, + }); }); it('should pass existingClock from rejection to superseded resolver (FIX: encryption conflict loop)', async () => { @@ -508,12 +691,13 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [], snapshotVectorClock: { snapshot: 1 }, } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -548,12 +732,13 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: remoteClocks, snapshotVectorClock: { snapshot: 1 }, } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -583,7 +768,7 @@ describe('RejectedOpsHandlerService', () => { const op = createOp({ id: 'op-1' }); opLogStoreSpy.getOpById.and.returnValue(Promise.resolve(mockEntry(op))); downloadCallback.and.callFake(async () => { - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); opLogStoreSpy.markRejected.and.resolveTo(); @@ -616,11 +801,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -671,11 +857,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); // First batch: should resolve all 4 ops (1 attempt counted) @@ -758,11 +945,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); // Send MAX_CONCURRENT_RESOLUTION_ATTEMPTS mixed batches @@ -831,11 +1019,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options: any) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -864,11 +1053,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options: any) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -899,11 +1089,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options: any) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -938,11 +1129,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options: any) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -1012,11 +1204,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); @@ -1036,11 +1229,12 @@ describe('RejectedOpsHandlerService', () => { downloadCallback.and.callFake(async (options) => { if (options?.forceFromSeq0) { return { + kind: 'completed', newOpsCount: 0, allOpClocks: [{ remoteClient: 2 }], } as DownloadResultForRejection; } - return { newOpsCount: 0 } as DownloadResultForRejection; + return { kind: 'completed', newOpsCount: 0 }; }); supersededOperationResolverSpy.resolveSupersededLocalOps.and.resolveTo(1); diff --git a/src/app/op-log/sync/rejected-ops-handler.service.ts b/src/app/op-log/sync/rejected-ops-handler.service.ts index 636d47ad9c..9bc9f436c1 100644 --- a/src/app/op-log/sync/rejected-ops-handler.service.ts +++ b/src/app/op-log/sync/rejected-ops-handler.service.ts @@ -22,12 +22,28 @@ export type { * a nested download) is surfaced via the SyncSessionValidationService latch * — the wrapper reads it once before deciding IN_SYNC vs ERROR. (#7330) */ -export interface RejectionHandlingResult { - /** Number of merged ops created from conflict resolution (these need to be uploaded) */ - mergedOpsCreated: number; - /** Number of operations that were permanently rejected (validation errors, etc.) */ - permanentRejectionCount: number; -} +export type RejectionHandlingResult = + | { + kind: 'completed'; + /** Number of merged ops created from conflict resolution (these need to be uploaded) */ + mergedOpsCreated: number; + /** Number of operations that were permanently rejected (validation errors, etc.) */ + permanentRejectionCount: number; + } + | { + /** User cancelled a nested SYNC_IMPORT conflict dialog. */ + kind: 'cancelled'; + }; + +type ConcurrentResolutionResult = + | { + kind: 'completed'; + mergedOpsCreated: number; + retryExceededCount: number; + } + | { + kind: 'cancelled'; + }; /** * Handles operations that were rejected by the server during upload. @@ -86,7 +102,11 @@ export class RejectedOpsHandlerService { if (rejectedOps.length === 0) { // No rejections = sync is healthy, reset resolution attempt counters this._resolutionAttemptsByEntity.clear(); - return { mergedOpsCreated: 0, permanentRejectionCount: 0 }; + return { + kind: 'completed', + mergedOpsCreated: 0, + permanentRejectionCount: 0, + }; } let mergedOpsCreated = 0; @@ -194,11 +214,15 @@ export class RejectedOpsHandlerService { concurrentModificationOps, downloadCallback, ); + if (result.kind === 'cancelled') { + return result; + } mergedOpsCreated = result.mergedOpsCreated; retryExceededCount = result.retryExceededCount; } return { + kind: 'completed', mergedOpsCreated, permanentRejectionCount: permanentlyRejectedOps.length + retryExceededCount, }; @@ -214,10 +238,7 @@ export class RejectedOpsHandlerService { existingClock?: VectorClock; }>, downloadCallback: DownloadCallback, - ): Promise<{ - mergedOpsCreated: number; - retryExceededCount: number; - }> { + ): Promise { let mergedOpsCreated = 0; // Check resolution attempt counts per entity to prevent infinite loops. @@ -258,7 +279,9 @@ export class RejectedOpsHandlerService { } } - // Mark exceeded-limit ops as permanently rejected to break the infinite loop + // Mark exceeded-limit ops as permanently rejected before downloading. Keeping + // them pending would let conflict resolution replace them with fresh ops and + // silently reset the per-entity retry budget. if (opsExceededRetries.length > 0) { OpLog.err( `RejectedOpsHandlerService: ${opsExceededRetries.length} ops exceeded max concurrent resolution attempts ` + @@ -270,7 +293,6 @@ export class RejectedOpsHandlerService { type: 'ERROR', msg: T.F.SYNC.S.CONFLICT_RESOLUTION_FAILED, }); - // Clean up tracking for rejected entities for (const item of opsExceededRetries) { this._resolutionAttemptsByEntity.delete(this._getEntityKey(item.op)); } @@ -278,6 +300,7 @@ export class RejectedOpsHandlerService { if (opsToResolve.length === 0) { return { + kind: 'completed', mergedOpsCreated: 0, retryExceededCount: opsExceededRetries.length, }; @@ -293,6 +316,10 @@ export class RejectedOpsHandlerService { // Validation failure (if any during the nested download) is on the // session-validation latch — wrapper reads it. (#7330) const downloadResult = await downloadCallback(); + if (downloadResult.kind === 'cancelled') { + this._rollbackResolutionAttempts(opsToResolve); + return { kind: 'cancelled' }; + } // Helper to check which ops are still pending, preserving existingClock from rejection const getStillPendingOps = async (): Promise< @@ -336,6 +363,10 @@ export class RejectedOpsHandlerService { ); const forceDownloadResult = await downloadCallback({ forceFromSeq0: true }); + if (forceDownloadResult.kind === 'cancelled') { + this._rollbackResolutionAttempts(opsToResolve); + return { kind: 'cancelled' }; + } // Use the clocks from force download to resolve superseded ops // Also merge in entity clocks from server rejection responses @@ -435,27 +466,32 @@ export class RejectedOpsHandlerService { // keeps re-downloading. Upgrade path if that becomes a problem: a separate // transient-failure budget via markFailed() (retryable, NOT terminal) with // backoff — never markRejected(). - const rolledBack = new Set(); - for (const { op } of opsToResolve) { - const entityKey = this._getEntityKey(op); - if (rolledBack.has(entityKey)) { - continue; - } - rolledBack.add(entityKey); - const attempts = this._resolutionAttemptsByEntity.get(entityKey) ?? 0; - if (attempts > 0) { - this._resolutionAttemptsByEntity.set(entityKey, attempts - 1); - } - } + this._rollbackResolutionAttempts(opsToResolve); throw e; } return { + kind: 'completed', mergedOpsCreated, retryExceededCount: opsExceededRetries.length, }; } + private _rollbackResolutionAttempts(ops: ReadonlyArray<{ op: Operation }>): void { + const rolledBack = new Set(); + for (const { op } of ops) { + const entityKey = this._getEntityKey(op); + if (rolledBack.has(entityKey)) { + continue; + } + rolledBack.add(entityKey); + const attempts = this._resolutionAttemptsByEntity.get(entityKey) ?? 0; + if (attempts > 0) { + this._resolutionAttemptsByEntity.set(entityKey, attempts - 1); + } + } + } + private _getEntityKey(op: Operation): string { const entityId = op.entityId || op.entityIds?.[0]; if (!entityId) { diff --git a/src/app/op-log/sync/server-migration.service.spec.ts b/src/app/op-log/sync/server-migration.service.spec.ts index d3d496500b..c4fe35f2fb 100644 --- a/src/app/op-log/sync/server-migration.service.spec.ts +++ b/src/app/op-log/sync/server-migration.service.spec.ts @@ -334,9 +334,10 @@ describe('ServerMigrationService', () => { } as any), ); - await service.handleServerMigration(defaultProvider); + const createdOpId = await service.handleServerMigration(defaultProvider); expect(opLogStoreSpy.append).toHaveBeenCalled(); + expect(createdOpId).toBe(opLogStoreSpy.append.calls.mostRecent().args[0].id); }); it('should proceed if non-entity sync state differs from defaults', async () => { diff --git a/src/app/op-log/sync/server-migration.service.ts b/src/app/op-log/sync/server-migration.service.ts index 2465c74055..d1799aa556 100644 --- a/src/app/op-log/sync/server-migration.service.ts +++ b/src/app/op-log/sync/server-migration.service.ts @@ -177,11 +177,12 @@ export class ServerMigrationService { * @param options - Optional configuration * @param options.skipServerEmptyCheck - If true, creates SYNC_IMPORT even if server has data. * Used for "USE_LOCAL" conflict resolution to force overwrite remote with local state. + * @returns The created SYNC_IMPORT operation ID, or undefined when creation was skipped. */ async handleServerMigration( syncProvider: OperationSyncCapable, options?: { skipServerEmptyCheck?: boolean; syncImportReason?: SyncImportReason }, - ): Promise { + ): Promise { // Double-check server is still empty (in case another client just uploaded) // This is called inside the upload lock, but network timing could still race // Skip this check when forcing upload (conflict resolution "USE_LOCAL") @@ -207,7 +208,7 @@ export class ServerMigrationService { // flushThenRunExclusive owns the flush→lock→recheck retry loop (bounded, so // continuous dispatch cannot livelock the migration; it re-triggers on the // next sync). - await this.writeFlushService.flushThenRunExclusive(async () => { + return this.writeFlushService.flushThenRunExclusive(async () => { // Get current full state from NgRx store (async to include archives from IndexedDB) // Cast to Record for validation compatibility let currentState: Record = @@ -310,6 +311,7 @@ export class ServerMigrationService { 'ServerMigrationService: Created SYNC_IMPORT operation for server migration. ' + 'Will be uploaded immediately via follow-up upload.', ); + return op.id; }); } diff --git a/src/app/op-log/sync/sync-import-conflict-coordinator.service.ts b/src/app/op-log/sync/sync-import-conflict-coordinator.service.ts index 0efa7b5342..f0224d9b25 100644 --- a/src/app/op-log/sync/sync-import-conflict-coordinator.service.ts +++ b/src/app/op-log/sync/sync-import-conflict-coordinator.service.ts @@ -8,12 +8,22 @@ import { import { OperationLogUploadService } from './operation-log-upload.service'; import { ServerMigrationService } from './server-migration.service'; import { SyncImportConflictDialogService } from './sync-import-conflict-dialog.service'; +import { + EncryptNoPasswordError, + ForceUploadFailedError, + ForceUploadPendingOpsError, +} from '../core/errors/sync-errors'; +import { OperationLogStoreService } from '../persistence/operation-log-store.service'; type SyncImportConflictActions = { - useLocal: () => Promise; + useLocal: () => Promise; useRemote: () => Promise; }; +export interface ForceUploadResult { + readonly hasUnresolvedOps: boolean; +} + @Injectable({ providedIn: 'root', }) @@ -21,6 +31,7 @@ export class SyncImportConflictCoordinatorService { private syncImportConflictDialogService = inject(SyncImportConflictDialogService); private serverMigrationService = inject(ServerMigrationService); private uploadService = inject(OperationLogUploadService); + private opLogStore = inject(OperationLogStoreService); async handleSyncImportConflict( dialogData: SyncImportConflictData, @@ -33,7 +44,11 @@ export class SyncImportConflictCoordinatorService { switch (resolution) { case 'USE_LOCAL': OpLog.normal(`${logPrefix}: User chose USE_LOCAL. Force uploading local state.`); - await actions.useLocal(); + if ((await actions.useLocal()).hasUnresolvedOps) { + throw new ForceUploadPendingOpsError( + 'Force upload succeeded with operations still pending.', + ); + } return 'USE_LOCAL'; case 'USE_REMOTE': OpLog.normal( @@ -48,21 +63,46 @@ export class SyncImportConflictCoordinatorService { } } - async forceUploadLocalState(syncProvider: OperationSyncCapable): Promise { + async forceUploadLocalState( + syncProvider: OperationSyncCapable, + ): Promise { OpLog.warn( 'SyncImportConflictCoordinatorService: Force uploading local state - creating SYNC_IMPORT to override remote.', ); - await this.serverMigrationService.handleServerMigration(syncProvider, { - skipServerEmptyCheck: true, - syncImportReason: 'FORCE_UPLOAD', - }); + const forceUploadOpId = await this.serverMigrationService.handleServerMigration( + syncProvider, + { + skipServerEmptyCheck: true, + syncImportReason: 'FORCE_UPLOAD', + }, + ); - await this.uploadService.uploadPendingOps(syncProvider, { + if (!forceUploadOpId) { + throw new ForceUploadFailedError( + 'Force upload failed because no SYNC_IMPORT was created.', + ); + } + + const uploadResult = await this.uploadService.uploadPendingOps(syncProvider, { skipPiggybackProcessing: true, isCleanSlate: true, }); + if (uploadResult.encryptionRequiredKeyMissing) { + throw new EncryptNoPasswordError( + 'Force upload requires an encryption key, but none is configured.', + ); + } + + const forceUploadEntry = await this.opLogStore.getOpById(forceUploadOpId); + if (!forceUploadEntry?.syncedAt) { + throw new ForceUploadFailedError( + 'Force upload failed because the SYNC_IMPORT was not accepted.', + ); + } + OpLog.normal('SyncImportConflictCoordinatorService: Force upload complete.'); + return { hasUnresolvedOps: uploadResult.rejectedCount > 0 }; } } diff --git a/src/app/op-log/sync/ws-triggered-download.service.spec.ts b/src/app/op-log/sync/ws-triggered-download.service.spec.ts index 1efd679ea9..170b0656f1 100644 --- a/src/app/op-log/sync/ws-triggered-download.service.spec.ts +++ b/src/app/op-log/sync/ws-triggered-download.service.spec.ts @@ -12,7 +12,11 @@ import { SyncSessionValidationService } from './sync-session-validation.service' import { SyncCycleGuardService } from './sync-cycle-guard.service'; import { SyncWrapperService } from '../../imex/sync/sync-wrapper.service'; import { AuthFailSPError, MissingCredentialsSPError } from '../sync-exports'; -import { IncompleteRemoteOperationsError } from '../core/errors/sync-errors'; +import { + ForceUploadFailedError, + ForceUploadPendingOpsError, + IncompleteRemoteOperationsError, +} from '../core/errors/sync-errors'; import { SnackService } from '../../core/snack/snack.service'; import { T } from '../../t.const'; @@ -270,6 +274,33 @@ describe('WsTriggeredDownloadService', () => { expect(mockSnackService.open).not.toHaveBeenCalled(); })); + it('should surface a force-upload failure raised by conflict resolution', fakeAsync(() => { + mockSyncService.downloadRemoteOps.and.rejectWith(new ForceUploadFailedError()); + + service.start(); + notification$.next({ latestSeq: 1 }); + tick(500); + flushMicrotasks(); + + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); + expect(mockSnackService.open).toHaveBeenCalledWith({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + })); + + it('should keep sync pending when force upload leaves unresolved ops', fakeAsync(() => { + mockSyncService.downloadRemoteOps.and.rejectWith(new ForceUploadPendingOpsError()); + + service.start(); + notification$.next({ latestSeq: 1 }); + tick(500); + flushMicrotasks(); + + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('UNKNOWN_OR_CHANGED'); + expect(mockSnackService.open).not.toHaveBeenCalled(); + })); + it('should be idempotent when start is called twice', fakeAsync(() => { service.start(); service.start(); diff --git a/src/app/op-log/sync/ws-triggered-download.service.ts b/src/app/op-log/sync/ws-triggered-download.service.ts index 3135c7e7ea..ab9e3c0d3e 100644 --- a/src/app/op-log/sync/ws-triggered-download.service.ts +++ b/src/app/op-log/sync/ws-triggered-download.service.ts @@ -11,7 +11,11 @@ import { SyncWrapperService } from '../../imex/sync/sync-wrapper.service'; import { lazyInject } from '../../util/lazy-inject'; import { SyncLog } from '../../core/log'; import { AuthFailSPError, MissingCredentialsSPError } from '../sync-exports'; -import { IncompleteRemoteOperationsError } from '../core/errors/sync-errors'; +import { + ForceUploadFailedError, + ForceUploadPendingOpsError, + IncompleteRemoteOperationsError, +} from '../core/errors/sync-errors'; import { SnackService } from '../../core/snack/snack.service'; import { T } from '../../t.const'; @@ -170,6 +174,20 @@ export class WsTriggeredDownloadService implements OnDestroy { this._providerManager.setSyncStatus('ERROR'); } } catch (err) { + if (err instanceof ForceUploadPendingOpsError) { + this._providerManager.setSyncStatus('UNKNOWN_OR_CHANGED'); + return; + } + + if (err instanceof ForceUploadFailedError) { + this._providerManager.setSyncStatus('ERROR'); + this._snackService.open({ + msg: T.F.SYNC.S.FORCE_UPLOAD_FAILED, + type: 'ERROR', + }); + return; + } + if (err instanceof IncompleteRemoteOperationsError) { SyncLog.err( 'WsTriggeredDownloadService: Remote operation application is incomplete', diff --git a/src/app/t.const.ts b/src/app/t.const.ts index 05cb3c1363..5226110a50 100644 --- a/src/app/t.const.ts +++ b/src/app/t.const.ts @@ -1667,6 +1667,7 @@ const T = { WARN_CLIENT_ID_REGENERATED: 'F.SYNC.S.WARN_CLIENT_ID_REGENERATED', ERROR_UNABLE_TO_READ_REMOTE_DATA: 'F.SYNC.S.ERROR_UNABLE_TO_READ_REMOTE_DATA', FINISH_DAY_SYNC_ERROR: 'F.SYNC.S.FINISH_DAY_SYNC_ERROR', + FORCE_UPLOAD_FAILED: 'F.SYNC.S.FORCE_UPLOAD_FAILED', FRESH_CLIENT_SYNC_CANCELLED: 'F.SYNC.S.FRESH_CLIENT_SYNC_CANCELLED', HYDRATION_FAILED: 'F.SYNC.S.HYDRATION_FAILED', IMPORTING: 'F.SYNC.S.IMPORTING', diff --git a/src/assets/i18n/en.json b/src/assets/i18n/en.json index b46b42b84c..49b2b03fe2 100644 --- a/src/assets/i18n/en.json +++ b/src/assets/i18n/en.json @@ -1619,6 +1619,7 @@ "WARN_CLIENT_ID_REGENERATED": "Sync device ID was invalid and has been regenerated. A sync conflict may occur on next sync.", "ERROR_UNABLE_TO_READ_REMOTE_DATA": "Error while syncing. Unable to read remote data. Maybe you enabled encryption and your local password does not match the one used to encrypt the remote data?", "FINISH_DAY_SYNC_ERROR": "Sync error while finishing day. Please try again.", + "FORCE_UPLOAD_FAILED": "Failed to overwrite server data. Your local changes remain available.", "FRESH_CLIENT_SYNC_CANCELLED": "Initial sync cancelled. Remote data was not applied.", "HYDRATION_FAILED": "Failed to load data. Please reload the app.", "IMPORTING": "Importing data",