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

fix(sync): preserve conflict cancellation and harden force overwrite (#8981)

Browse source 
* 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
This commit is contained in:
Johannes Millan 2026-07-13 21:54:07 +02:00 • committed by GitHub
parent bd67174863
commit 55d3490e19
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 1054 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<string>,
): Promise<UploadResult[]> {
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';

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

@ -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> = {}): 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<SyncConfig> = { 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();
},
);
});

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

@ -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[] = [

93
src/app/imex/sync/sync-wrapper.service.spec.ts
View file

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

37
src/app/imex/sync/sync-wrapper.service.ts
View file

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

8
src/app/op-log/core/errors/sync-errors.ts
View file

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

16
src/app/op-log/core/types/sync-results.types.ts
View file

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

35
src/app/op-log/sync/immediate-upload.service.spec.ts
View file

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

20
src/app/op-log/sync/immediate-upload.service.ts
View file

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

150
src/app/op-log/sync/operation-log-sync.service.spec.ts
View file

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

20
src/app/op-log/sync/operation-log-sync.service.ts
View file

@ -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<void> {
await this.syncImportConflictCoordinator.forceUploadLocalState(syncProvider);
async forceUploadLocalState(
syncProvider: OperationSyncCapable,
): Promise<ForceUploadResult> {
return this.syncImportConflictCoordinator.forceUploadLocalState(syncProvider);
}
/**

1
src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts
View file

@ -234,6 +234,7 @@ describe('OperationLogSyncService + OperationLogUploadService — piggyback seq
['handleRejectedOps'],
);
rejectedOpsHandlerServiceSpy.handleRejectedOps.and.resolveTo({
kind: 'completed',
mergedOpsCreated: 0,
permanentRejectionCount: 0,
});

16
src/app/op-log/sync/operation-log-upload.service.spec.ts
View file

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

4
src/app/op-log/sync/operation-log-upload.service.ts
View file

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

234
src/app/op-log/sync/rejected-ops-handler.service.spec.ts
View file

@ -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<string, ReturnType<typeof mockEntry>>();
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);

86
src/app/op-log/sync/rejected-ops-handler.service.ts
View file

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

3
src/app/op-log/sync/server-migration.service.spec.ts
View file

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

6
src/app/op-log/sync/server-migration.service.ts
View file

@ -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<void> {
): Promise<string | undefined> {
// 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<string, unknown> =
@ -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;
});
}

56
src/app/op-log/sync/sync-import-conflict-coordinator.service.ts
View file

@ -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<void>;
useLocal: () => Promise<ForceUploadResult>;
useRemote: () => Promise<void>;
};
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<void> {
async forceUploadLocalState(
syncProvider: OperationSyncCapable,
): Promise<ForceUploadResult> {
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 };
}
}

33
src/app/op-log/sync/ws-triggered-download.service.spec.ts
View file

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

20
src/app/op-log/sync/ws-triggered-download.service.ts
View file

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

1
src/app/t.const.ts
View file

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

1
src/assets/i18n/en.json
View file

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