mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
fix(sync): handle initial provider setup safely
This commit is contained in:
Johannes Millan
parent
bfdc63e0c6
commit
eaa15fe575
9 changed files with 571 additions and 86 deletions
|
|
@ -105,6 +105,6 @@ flowchart TD
|
|||
**Notes:**
|
||||
|
||||
- The `Enter Password` and `Decrypt Error` dialogs correspond to `DecryptNoPasswordError` and `DecryptError` respectively — they are distinct components with different options.
|
||||
- `IMPORT_CONFLICT` uses pending ops only, not already-synced store contents (`_hasMeaningfulPendingOps()`). Every pending op is protected except onboarding example-task creates. This includes GLOBAL_CONFIG changes before first sync, non-task entities, and MIGRATION/RECOVERY genesis batches. PASSWORD_CHANGED SYNC_IMPORTs without pending ops still fall through to silent acceptance — the data is identical, only the encryption changed. Including already-synced store contents would let an old client pick `USE_LOCAL` and force-upload stale pre-import state, rolling back the remote import for everyone.
|
||||
- `IMPORT_CONFLICT` uses pending ops only, not already-synced store contents (`_hasMeaningfulPendingOps()`). Every pending op is protected except onboarding example-task creates and the never-synced `GLOBAL_CONFIG:sync` provider-setup write. Other config sections, non-task entities, and MIGRATION/RECOVERY full-state batches remain protected. PASSWORD_CHANGED SYNC_IMPORTs without meaningful pending ops still fall through to silent acceptance — the data is identical, only the encryption changed. Including already-synced store contents would let an old client pick `USE_LOCAL` and force-upload stale pre-import state, rolling back the remote import for everyone.
|
||||
- LWW tie-breaking: on equal timestamps, remote wins (server-authoritative). `moveToArchive` operations always win regardless of timestamp.
|
||||
- Re-download retry limit: max 3 resolution attempts per entity (`MAX_CONCURRENT_RESOLUTION_ATTEMPTS`); if exceeded, ops are permanently rejected.
|
||||
|
|
|
|||
|
|
@ -165,9 +165,9 @@ Comprehensive spec of all scenarios that can occur during SuperSync synchronizat
|
|||
**Expected:**
|
||||
|
||||
1. Download detects remote snapshot (file-based sync path)
|
||||
2. Treat every pending op as meaningful except onboarding example-task creates. GLOBAL_CONFIG writes remain protected even before the first completed sync because the sync-section payload can also contain user preferences. This also protects non-task entities and recovered MIGRATION/RECOVERY genesis batches.
|
||||
2. Treat every pending op as meaningful except onboarding example-task creates and, before the first completed sync only, the `GLOBAL_CONFIG:sync` setup write needed to configure the provider. Other config sections, non-task entities, and MIGRATION/RECOVERY full-state batches remain protected.
|
||||
3. If meaningful → throw `LocalDataConflictError` → full conflict dialog
|
||||
4. If only the explicitly discardable startup ops remain → proceed without dialog
|
||||
4. If only the explicitly discardable startup ops remain → proceed without dialog and reject those ops locally so they cannot replay after the imported state
|
||||
|
||||
**Note:** This op-content check only applies to the file-based snapshot path. For SuperSync (incremental ops path), the fresh client check uses `_hasMeaningfulStoreData()` (store-based check) instead.
|
||||
|
||||
|
|
@ -198,7 +198,7 @@ Comprehensive spec of all scenarios that can occur during SuperSync synchronizat
|
|||
**Expected:**
|
||||
|
||||
1. Download batch contains SYNC_IMPORT
|
||||
2. Check pending local ops. Any pending work triggers the dialog except onboarding example-task creates. GLOBAL_CONFIG changes are protected even before the first completed sync because the sync-section payload can carry user preferences.
|
||||
2. Check pending local ops. Any pending work triggers the dialog except onboarding example-task creates and the never-synced `GLOBAL_CONFIG:sync` provider-setup write. Other GLOBAL_CONFIG sections remain protected.
|
||||
3. **Show conflict dialog BEFORE processing** with `scenario: 'INCOMING_IMPORT'` and `syncImportReason`
|
||||
4. USE_LOCAL → `forceUploadLocalState()` (overrides remote with local data)
|
||||
5. USE_REMOTE → `forceDownloadRemoteState()` (clears local ops, downloads from seq 0)
|
||||
|
|
@ -254,12 +254,12 @@ Comprehensive spec of all scenarios that can occur during SuperSync synchronizat
|
|||
|
||||
1. Upload completes → server returns piggybacked ops containing SYNC_IMPORT
|
||||
2. Check for SYNC_IMPORT in piggybacked ops BEFORE `processRemoteOps()`
|
||||
3. If found AND `_hasMeaningfulPendingOps()` = true (all pending work except onboarding example-task creates):
|
||||
3. If found AND `_hasMeaningfulPendingOps()` = true (all pending work except onboarding example-task creates and the never-synced `GLOBAL_CONFIG:sync` provider-setup write):
|
||||
- **Show conflict dialog** with `scenario: 'INCOMING_IMPORT'` and `syncImportReason` from the piggybacked op
|
||||
- USE_LOCAL → `forceUploadLocalState()` (overrides remote)
|
||||
- USE_REMOTE → `forceDownloadRemoteState()` (clears local, downloads from seq 0)
|
||||
- CANCEL → return with `cancelled: true`, callers skip post-upload logic
|
||||
4. If no meaningful pending ops → `processRemoteOps()` applies silently (no dialog) regardless of whether the NgRx store already has user data — that data was already synced and the SYNC_IMPORT is the new authoritative state.
|
||||
4. If no meaningful pending ops → `processRemoteOps()` applies silently (no dialog), then reject live discardable startup ops only after processing succeeds. Existing NgRx store data does not trigger a dialog because it was already synced and the SYNC_IMPORT is the new authoritative state.
|
||||
|
||||
**Mirrors the download path (D.1 / D.2):** the gate is unsynced pending changes, not store contents. Prompting on already-synced store data would let an old client roll back the remote import via USE_LOCAL.
|
||||
|
||||
|
|
|
|||
|
|
@ -12,7 +12,10 @@ import { ValidateStateService } from '../validation/validate-state.service';
|
|||
import { SyncSessionValidationService } from './sync-session-validation.service';
|
||||
import { RepairOperationService } from '../validation/repair-operation.service';
|
||||
import { OperationLogUploadService } from './operation-log-upload.service';
|
||||
import { OperationLogDownloadService } from './operation-log-download.service';
|
||||
import {
|
||||
DownloadResult,
|
||||
OperationLogDownloadService,
|
||||
} from './operation-log-download.service';
|
||||
import { LockService } from './lock.service';
|
||||
import { OperationLogCompactionService } from '../persistence/operation-log-compaction.service';
|
||||
import { SyncImportFilterService } from './sync-import-filter.service';
|
||||
|
|
@ -45,6 +48,7 @@ import { OperationSyncCapable } from '../sync-providers/provider.interface';
|
|||
import { selectSyncConfig } from '../../features/config/store/global-config.reducer';
|
||||
import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const';
|
||||
import { SyncProviderId } from '../sync-providers/provider.const';
|
||||
import { stripLocalOnlySyncSettingsFromAppData } from '../../features/config/local-only-sync-settings.util';
|
||||
|
||||
describe('OperationLogSyncService', () => {
|
||||
let service: OperationLogSyncService;
|
||||
|
|
@ -64,6 +68,24 @@ describe('OperationLogSyncService', () => {
|
|||
let operationApplierSpy: jasmine.SpyObj<OperationApplierService>;
|
||||
let operationLogEffectsSpy: jasmine.SpyObj<OperationLogEffects>;
|
||||
|
||||
const createProviderSetupEntry = (): OperationLogEntry => ({
|
||||
seq: 1,
|
||||
op: {
|
||||
id: 'sync-provider-setup',
|
||||
clientId: 'client-A',
|
||||
actionType: ActionType.GLOBAL_CONFIG_UPDATE_SECTION,
|
||||
opType: OpType.Update,
|
||||
entityType: 'GLOBAL_CONFIG',
|
||||
entityId: 'sync',
|
||||
payload: { sectionKey: 'sync' },
|
||||
vectorClock: { clientA: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
},
|
||||
appliedAt: Date.now(),
|
||||
source: 'local',
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
snackServiceSpy = jasmine.createSpyObj('SnackService', [
|
||||
'open',
|
||||
|
|
@ -1669,10 +1691,8 @@ describe('OperationLogSyncService', () => {
|
|||
source: 'local',
|
||||
});
|
||||
|
||||
const fileSnapshotDownloadResult = {
|
||||
const fileSnapshotDownloadResult: DownloadResult = {
|
||||
newOps: [],
|
||||
hasMore: false,
|
||||
latestSeq: 0,
|
||||
needsFullStateUpload: false,
|
||||
success: true,
|
||||
providerMode: 'fileSnapshotOps',
|
||||
|
|
@ -1682,6 +1702,32 @@ describe('OperationLogSyncService', () => {
|
|||
latestServerSeq: 1,
|
||||
};
|
||||
|
||||
it('silently adopts a file snapshot and rejects the never-synced provider setup op', async () => {
|
||||
const setupEntry = createProviderSetupEntry();
|
||||
opLogStoreSpy.hasSyncedOps.and.resolveTo(false);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([setupEntry]);
|
||||
|
||||
const syncHydrationServiceSpy = TestBed.inject(
|
||||
SyncHydrationService,
|
||||
) as jasmine.SpyObj<SyncHydrationService>;
|
||||
syncHydrationServiceSpy.hydrateFromRemoteSync.and.resolveTo();
|
||||
downloadServiceSpy.downloadRemoteOps.and.resolveTo(fileSnapshotDownloadResult);
|
||||
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
await expectAsync(
|
||||
service.downloadRemoteOps(mockProvider, { isNeverSynced: true }),
|
||||
).toBeResolved();
|
||||
expect(syncHydrationServiceSpy.hydrateFromRemoteSync).toHaveBeenCalled();
|
||||
expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([
|
||||
'sync-provider-setup',
|
||||
]);
|
||||
});
|
||||
|
||||
it('does NOT throw LocalDataConflictError when store + pending ops contain only example tasks (#7985)', async () => {
|
||||
opLogStoreSpy.getUnsynced.and.returnValue(
|
||||
Promise.resolve([exampleCreateEntry('ex-task-1')]),
|
||||
|
|
@ -1700,7 +1746,7 @@ describe('OperationLogSyncService', () => {
|
|||
syncHydrationServiceSpy.hydrateFromRemoteSync.and.resolveTo();
|
||||
|
||||
downloadServiceSpy.downloadRemoteOps.and.returnValue(
|
||||
Promise.resolve(fileSnapshotDownloadResult as any),
|
||||
Promise.resolve(fileSnapshotDownloadResult),
|
||||
);
|
||||
|
||||
const mockProvider = {
|
||||
|
|
@ -1747,7 +1793,7 @@ describe('OperationLogSyncService', () => {
|
|||
);
|
||||
|
||||
downloadServiceSpy.downloadRemoteOps.and.returnValue(
|
||||
Promise.resolve(fileSnapshotDownloadResult as any),
|
||||
Promise.resolve(fileSnapshotDownloadResult),
|
||||
);
|
||||
|
||||
const mockProvider = {
|
||||
|
|
@ -1774,7 +1820,7 @@ describe('OperationLogSyncService', () => {
|
|||
} as any);
|
||||
|
||||
downloadServiceSpy.downloadRemoteOps.and.returnValue(
|
||||
Promise.resolve(fileSnapshotDownloadResult as any),
|
||||
Promise.resolve(fileSnapshotDownloadResult),
|
||||
);
|
||||
|
||||
const mockProvider = {
|
||||
|
|
@ -2656,7 +2702,7 @@ describe('OperationLogSyncService', () => {
|
|||
const mockProvider = {
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
|
||||
} as any;
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
await service.forceUploadLocalState(mockProvider);
|
||||
|
||||
|
|
@ -3230,6 +3276,55 @@ describe('OperationLogSyncService', () => {
|
|||
expect(opLogStoreSpy.runRemoteStateReplacement).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should restore device-local sync settings before validating a file snapshot', async () => {
|
||||
const mockStore = TestBed.inject(MockStore);
|
||||
mockStore.overrideSelector(selectSyncConfig, {
|
||||
...DEFAULT_GLOBAL_CONFIG.sync,
|
||||
syncProvider: SyncProviderId.WebDAV,
|
||||
isEnabled: true,
|
||||
isEncryptionEnabled: true,
|
||||
syncInterval: 23,
|
||||
isManualSyncOnly: true,
|
||||
});
|
||||
mockStore.refreshState();
|
||||
const wireSnapshot = stripLocalOnlySyncSettingsFromAppData({
|
||||
globalConfig: DEFAULT_GLOBAL_CONFIG,
|
||||
});
|
||||
downloadServiceSpy.downloadRemoteOps.and.resolveTo({
|
||||
newOps: [],
|
||||
needsFullStateUpload: false,
|
||||
success: true,
|
||||
providerMode: 'fileSnapshotOps',
|
||||
failedFileCount: 0,
|
||||
snapshotState: wireSnapshot,
|
||||
latestServerSeq: 1,
|
||||
});
|
||||
validateStateServiceSpy.validateAndRepair.and.resolveTo({
|
||||
isValid: true,
|
||||
wasRepaired: false,
|
||||
});
|
||||
const mockProvider = {
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
await service.forceDownloadRemoteState(mockProvider);
|
||||
|
||||
expect(validateStateServiceSpy.validateAndRepair).toHaveBeenCalledOnceWith(
|
||||
jasmine.objectContaining({
|
||||
globalConfig: jasmine.objectContaining({
|
||||
sync: jasmine.objectContaining({
|
||||
syncProvider: SyncProviderId.WebDAV,
|
||||
isEnabled: true,
|
||||
isEncryptionEnabled: true,
|
||||
syncInterval: 23,
|
||||
isManualSyncOnly: true,
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('should NOT advance the cursor past 0 when replay blocks on a migration failure', async () => {
|
||||
downloadServiceSpy.downloadRemoteOps.and.resolveTo({
|
||||
newOps: [makeRemoteOp()],
|
||||
|
|
@ -4025,6 +4120,81 @@ describe('OperationLogSyncService', () => {
|
|||
expect(result.kind).toBe('ops_processed');
|
||||
});
|
||||
|
||||
it('should discard initial provider setup only after the incoming import commits', async () => {
|
||||
const incomingSyncImport = createIncomingSyncImport();
|
||||
const setupEntry = createProviderSetupEntry();
|
||||
downloadServiceSpy.downloadRemoteOps.and.resolveTo({
|
||||
newOps: [incomingSyncImport],
|
||||
success: true,
|
||||
providerMode: 'superSyncOps',
|
||||
failedFileCount: 0,
|
||||
latestServerSeq: 42,
|
||||
});
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([setupEntry]);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp: true,
|
||||
});
|
||||
const mockProvider = {
|
||||
supportsOperationSync: true,
|
||||
setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
const result = await service.downloadRemoteOps(mockProvider, {
|
||||
isNeverSynced: true,
|
||||
});
|
||||
|
||||
expect(result.kind).toBe('blocked_incompatible');
|
||||
expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled();
|
||||
expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled();
|
||||
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp: true,
|
||||
committedFullStateOpIds: [incomingSyncImport.id],
|
||||
});
|
||||
|
||||
const committedPrefixResult = await service.downloadRemoteOps(mockProvider, {
|
||||
isNeverSynced: true,
|
||||
});
|
||||
|
||||
expect(committedPrefixResult.kind).toBe('blocked_incompatible');
|
||||
expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([setupEntry.op.id]);
|
||||
expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled();
|
||||
|
||||
const processingError = new Error('deferred drain failed');
|
||||
opLogStoreSpy.markRejected.calls.reset();
|
||||
opLogStoreSpy.getOpById.and.resolveTo(undefined);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.rejectWith(processingError);
|
||||
|
||||
await expectAsync(
|
||||
service.downloadRemoteOps(mockProvider, { isNeverSynced: true }),
|
||||
).toBeRejectedWith(processingError);
|
||||
expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled();
|
||||
|
||||
for (const applicationStatus of ['applied', 'archive_pending', 'failed'] as const) {
|
||||
opLogStoreSpy.markRejected.calls.reset();
|
||||
opLogStoreSpy.getOpById.and.resolveTo({
|
||||
seq: 2,
|
||||
op: incomingSyncImport,
|
||||
appliedAt: Date.now(),
|
||||
source: 'remote',
|
||||
applicationStatus,
|
||||
});
|
||||
|
||||
await expectAsync(
|
||||
service.downloadRemoteOps(mockProvider, { isNeverSynced: true }),
|
||||
).toBeRejectedWith(processingError);
|
||||
expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([setupEntry.op.id]);
|
||||
}
|
||||
});
|
||||
|
||||
it('should show conflict dialog for incoming SYNC_IMPORT when client has pending meaningful ops', async () => {
|
||||
const incomingSyncImport = createIncomingSyncImport();
|
||||
|
||||
|
|
@ -4345,7 +4515,7 @@ describe('OperationLogSyncService', () => {
|
|||
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
} as any;
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider);
|
||||
|
||||
|
|
@ -4405,7 +4575,7 @@ describe('OperationLogSyncService', () => {
|
|||
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
} as any;
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider);
|
||||
|
||||
|
|
@ -4463,6 +4633,113 @@ describe('OperationLogSyncService', () => {
|
|||
expect(result.kind).toBe('completed');
|
||||
});
|
||||
|
||||
it('should reject a live provider setup op after silently applying a piggybacked import on initial sync', async () => {
|
||||
const piggybackedSyncImport: Operation = {
|
||||
id: 'import-1',
|
||||
clientId: 'client-B',
|
||||
actionType: ActionType.LOAD_ALL_DATA,
|
||||
opType: OpType.SyncImport,
|
||||
entityType: 'ALL',
|
||||
payload: {},
|
||||
vectorClock: { clientB: 5 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
const setupEntry = createProviderSetupEntry();
|
||||
|
||||
uploadServiceSpy.uploadPendingOps.and.resolveTo({
|
||||
uploadedCount: 1,
|
||||
piggybackedOps: [piggybackedSyncImport],
|
||||
rejectedCount: 0,
|
||||
rejectedOps: [],
|
||||
selectedPendingOps: [setupEntry],
|
||||
});
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([setupEntry]);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp: false,
|
||||
committedFullStateOpIds: [piggybackedSyncImport.id],
|
||||
});
|
||||
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider, {
|
||||
isNeverSynced: true,
|
||||
});
|
||||
|
||||
expect(
|
||||
syncImportConflictDialogServiceSpy.showConflictDialog,
|
||||
).not.toHaveBeenCalled();
|
||||
expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([
|
||||
'sync-provider-setup',
|
||||
]);
|
||||
expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([
|
||||
piggybackedSyncImport,
|
||||
]);
|
||||
expect(result.kind).toBe('completed');
|
||||
});
|
||||
|
||||
it('should keep the initial provider setup op pending when piggybacked import processing is blocked', async () => {
|
||||
const piggybackedSyncImport: Operation = {
|
||||
id: 'import-1',
|
||||
clientId: 'client-B',
|
||||
actionType: ActionType.LOAD_ALL_DATA,
|
||||
opType: OpType.SyncImport,
|
||||
entityType: 'ALL',
|
||||
payload: {},
|
||||
vectorClock: { clientB: 5 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
const setupEntry = createProviderSetupEntry();
|
||||
uploadServiceSpy.uploadPendingOps.and.resolveTo({
|
||||
uploadedCount: 1,
|
||||
piggybackedOps: [piggybackedSyncImport],
|
||||
rejectedCount: 0,
|
||||
rejectedOps: [],
|
||||
selectedPendingOps: [setupEntry],
|
||||
});
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([setupEntry]);
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp: true,
|
||||
});
|
||||
const mockProvider = {
|
||||
isReady: () => Promise.resolve(true),
|
||||
} as unknown as OperationSyncCapable;
|
||||
|
||||
const result = await service.uploadPendingOps(mockProvider, {
|
||||
isNeverSynced: true,
|
||||
});
|
||||
|
||||
expect(result.kind).toBe('blocked_incompatible');
|
||||
expect(opLogStoreSpy.markRejected).not.toHaveBeenCalled();
|
||||
|
||||
remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({
|
||||
localWinOpsCreated: 0,
|
||||
allOpsFilteredBySyncImport: false,
|
||||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp: true,
|
||||
committedFullStateOpIds: [piggybackedSyncImport.id],
|
||||
});
|
||||
|
||||
const committedPrefixResult = await service.uploadPendingOps(mockProvider, {
|
||||
isNeverSynced: true,
|
||||
});
|
||||
|
||||
expect(committedPrefixResult.kind).toBe('blocked_incompatible');
|
||||
expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([setupEntry.op.id]);
|
||||
});
|
||||
|
||||
it('should flush again before checking piggybacked SYNC_IMPORT conflicts', async () => {
|
||||
const piggybackedSyncImport: Operation = {
|
||||
id: 'import-1',
|
||||
|
|
|
|||
|
|
@ -61,11 +61,16 @@ import { selectSyncConfig } from '../../features/config/store/global-config.redu
|
|||
import {
|
||||
applyLocalOnlySyncSettingsToAppData,
|
||||
LocalOnlySyncSettings,
|
||||
stripLocalOnlySyncSettingsFromAppData,
|
||||
} from '../../features/config/local-only-sync-settings.util';
|
||||
import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const';
|
||||
import { OperationApplierService } from '../apply/operation-applier.service';
|
||||
import { processDeferredActions } from './process-deferred-actions-flush.util';
|
||||
|
||||
type RemoteOpsProcessingResult = Awaited<
|
||||
ReturnType<RemoteOpsProcessingService['processRemoteOps']>
|
||||
>;
|
||||
|
||||
/**
|
||||
* Orchestrates synchronization of the Operation Log with remote storage.
|
||||
*
|
||||
|
|
@ -268,6 +273,8 @@ export class OperationLogSyncService {
|
|||
};
|
||||
|
||||
if (result.piggybackedOps.length > 0) {
|
||||
let startupOpIdsToDiscard: string[] = [];
|
||||
let startupCleanupFullStateOpId: string | undefined;
|
||||
// Check for piggybacked SYNC_IMPORT — mirrors the download path check (lines 552-604).
|
||||
// Without this, a SYNC_IMPORT from another client arriving as a piggybacked op
|
||||
// would silently replace local state via processRemoteOps().
|
||||
|
|
@ -323,7 +330,8 @@ export class OperationLogSyncService {
|
|||
// against the import (SyncImportFilterService). Only reachable in the narrow window
|
||||
// where example tasks are created on a still-empty server and uploaded just as a
|
||||
// remote import arrives; afterInitialSyncDoneStrict$ shrinks it further.
|
||||
await this._discardExampleTaskOps(piggybackedConflict.discardablePendingOpIds);
|
||||
startupOpIdsToDiscard = piggybackedConflict.discardablePendingOpIds;
|
||||
startupCleanupFullStateOpId = fullStateOp.id;
|
||||
OpLog.normal(
|
||||
`OperationLogSyncService: Accepting piggybacked ${fullStateOp.opType} from client ` +
|
||||
`${fullStateOp.clientId} without conflict dialog; ` +
|
||||
|
|
@ -332,8 +340,10 @@ export class OperationLogSyncService {
|
|||
}
|
||||
}
|
||||
|
||||
const processResult = await this.remoteOpsProcessingService.processRemoteOps(
|
||||
const processResult = await this._processRemoteOpsWithStartupCleanup(
|
||||
result.piggybackedOps,
|
||||
startupCleanupFullStateOpId,
|
||||
startupOpIdsToDiscard,
|
||||
);
|
||||
localWinOpsCreated = processResult.localWinOpsCreated;
|
||||
// Validation failure (if any) is on the session-validation latch.
|
||||
|
|
@ -559,9 +569,9 @@ export class OperationLogSyncService {
|
|||
const hasLocalChanges = unsyncedOps.length > 0;
|
||||
|
||||
// Collected here, applied AFTER hydrateFromRemoteSync succeeds so a
|
||||
// hydration failure doesn't permanently drop the pending example-create
|
||||
// ops while leaving the user without the remote snapshot.
|
||||
let exampleTaskOpIdsToDiscard: string[] = [];
|
||||
// hydration failure doesn't permanently drop discardable startup ops
|
||||
// while leaving the user without the remote snapshot.
|
||||
let startupOpIdsToDiscard: string[] = [];
|
||||
|
||||
if (hasLocalChanges) {
|
||||
// Throw LocalDataConflictError if unsynced ops contain meaningful user data
|
||||
|
|
@ -585,12 +595,23 @@ export class OperationLogSyncService {
|
|||
.map((entry) => entry.op.entityId)
|
||||
.filter((id): id is string => id !== undefined),
|
||||
);
|
||||
const exampleTaskOpIds = exampleTaskEntries.map((entry) => entry.op.id);
|
||||
// Nothing from this sync is persisted yet, so this live read reflects
|
||||
// whether the client completed a prior sync cycle.
|
||||
const isNeverSyncedAtSyncStart =
|
||||
options?.isNeverSynced ?? !(await this.opLogStore.hasSyncedOps());
|
||||
const pendingOpClassification = {
|
||||
hasCompletedInitialSync: !isNeverSyncedAtSyncStart,
|
||||
};
|
||||
const discardableStartupOpIds =
|
||||
this.syncImportConflictGateService.getDiscardablePendingOpIds(
|
||||
unsyncedOps,
|
||||
pendingOpClassification,
|
||||
);
|
||||
const hasMeaningfulUserData =
|
||||
this.syncImportConflictGateService.hasMeaningfulPendingOps(unsyncedOps) ||
|
||||
this.syncLocalStateService.hasMeaningfulStoreData(exampleTaskIds);
|
||||
this.syncImportConflictGateService.hasMeaningfulPendingOps(
|
||||
unsyncedOps,
|
||||
pendingOpClassification,
|
||||
) || this.syncLocalStateService.hasMeaningfulStoreData(exampleTaskIds);
|
||||
|
||||
if (hasMeaningfulUserData) {
|
||||
// SPAP-9: before surfacing the binary USE_LOCAL/USE_REMOTE dialog, use
|
||||
|
|
@ -674,15 +695,15 @@ export class OperationLogSyncService {
|
|||
// gate === 'apply-snapshot': the remote snapshot strictly dominates the
|
||||
// local clock, so local holds nothing the snapshot lacks. Adopt the
|
||||
// snapshot without a dialog by falling through to hydration below.
|
||||
exampleTaskOpIdsToDiscard = exampleTaskOpIds;
|
||||
startupOpIdsToDiscard = discardableStartupOpIds;
|
||||
OpLog.normal(
|
||||
'OperationLogSyncService: Remote snapshot strictly ahead of local clock — ' +
|
||||
'applying snapshot without conflict dialog.',
|
||||
);
|
||||
} else {
|
||||
// Defer the markRejected call until hydration has succeeded — see
|
||||
// the declaration of exampleTaskOpIdsToDiscard above for rationale.
|
||||
exampleTaskOpIdsToDiscard = exampleTaskOpIds;
|
||||
// the declaration of startupOpIdsToDiscard above for rationale.
|
||||
startupOpIdsToDiscard = discardableStartupOpIds;
|
||||
// Only system/config ops AND no meaningful store data - proceed with download
|
||||
OpLog.normal(
|
||||
`OperationLogSyncService: Client has ${unsyncedOps.length} unsynced ops but no meaningful user data. ` +
|
||||
|
|
@ -747,10 +768,10 @@ export class OperationLogSyncService {
|
|||
);
|
||||
|
||||
// Now that the remote snapshot is applied, it's safe to drop the
|
||||
// example-create ops we previously decided were obsolete. Doing this
|
||||
// startup ops we previously decided were obsolete. Doing this
|
||||
// after hydration ensures a hydration failure leaves the queue intact
|
||||
// so the next attempt can retry.
|
||||
await this._discardExampleTaskOps(exampleTaskOpIdsToDiscard);
|
||||
await this._discardStartupOps(startupOpIdsToDiscard);
|
||||
|
||||
// CRITICAL FIX: Write recentOps to IndexedDB after snapshot hydration.
|
||||
// File-based providers return ALL recentOps on every download, relying on
|
||||
|
|
@ -889,6 +910,8 @@ export class OperationLogSyncService {
|
|||
isNeverSynced: options?.isNeverSynced,
|
||||
},
|
||||
);
|
||||
let startupOpIdsToDiscard: string[] = [];
|
||||
let startupCleanupFullStateOpId: string | undefined;
|
||||
if (incomingConflict.fullStateOp) {
|
||||
const { fullStateOp, pendingOps, dialogData } = incomingConflict;
|
||||
// Existing synced store data is not a conflict here. Prompt only when
|
||||
|
|
@ -915,7 +938,8 @@ export class OperationLogSyncService {
|
|||
// the session-validation latch — wrapper reads it. (#7330)
|
||||
return { kind: 'no_new_ops' };
|
||||
} else {
|
||||
await this._discardExampleTaskOps(incomingConflict.discardablePendingOpIds);
|
||||
startupOpIdsToDiscard = incomingConflict.discardablePendingOpIds;
|
||||
startupCleanupFullStateOpId = fullStateOp.id;
|
||||
OpLog.normal(
|
||||
`OperationLogSyncService: Accepting incoming ${fullStateOp.opType} from client ` +
|
||||
`${fullStateOp.clientId} without conflict dialog; ` +
|
||||
|
|
@ -924,8 +948,10 @@ export class OperationLogSyncService {
|
|||
}
|
||||
}
|
||||
|
||||
const processResult = await this.remoteOpsProcessingService.processRemoteOps(
|
||||
const processResult = await this._processRemoteOpsWithStartupCleanup(
|
||||
result.newOps,
|
||||
startupCleanupFullStateOpId,
|
||||
startupOpIdsToDiscard,
|
||||
);
|
||||
|
||||
if (processResult.blockedByIncompatibleOp) {
|
||||
|
|
@ -1010,17 +1036,74 @@ export class OperationLogSyncService {
|
|||
};
|
||||
}
|
||||
|
||||
private async _processRemoteOpsWithStartupCleanup(
|
||||
remoteOps: Operation[],
|
||||
fullStateOpId: string | undefined,
|
||||
startupOpIds: string[],
|
||||
): Promise<RemoteOpsProcessingResult> {
|
||||
try {
|
||||
const result = await this.remoteOpsProcessingService.processRemoteOps(remoteOps);
|
||||
await this._discardStartupOpsIfFullStateCommitted(
|
||||
fullStateOpId,
|
||||
startupOpIds,
|
||||
result.committedFullStateOpIds,
|
||||
);
|
||||
return result;
|
||||
} catch (error) {
|
||||
try {
|
||||
// The reducer/apply transaction can commit the full-state op before a
|
||||
// later validation or deferred-action drain throws. Query persistence
|
||||
// so obsolete startup ops cannot replay after an already-applied import.
|
||||
await this._discardStartupOpsIfFullStateCommitted(
|
||||
fullStateOpId,
|
||||
startupOpIds,
|
||||
[],
|
||||
true,
|
||||
);
|
||||
} catch (cleanupError) {
|
||||
// Preserve the primary processing error. A later retry can re-check and
|
||||
// clean up once persistence is available again.
|
||||
OpLog.err(
|
||||
'OperationLogSyncService: Failed to verify startup-op cleanup after remote processing error.',
|
||||
{ name: (cleanupError as Error | undefined)?.name },
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async _discardStartupOpsIfFullStateCommitted(
|
||||
fullStateOpId: string | undefined,
|
||||
startupOpIds: string[],
|
||||
committedFullStateOpIds: string[] = [],
|
||||
acceptReducerCommittedFailureStatus: boolean = false,
|
||||
): Promise<void> {
|
||||
if (!fullStateOpId || startupOpIds.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const applicationStatus = (await this.opLogStore.getOpById(fullStateOpId))
|
||||
?.applicationStatus;
|
||||
const isCommitted =
|
||||
committedFullStateOpIds.includes(fullStateOpId) ||
|
||||
applicationStatus === 'applied' ||
|
||||
(acceptReducerCommittedFailureStatus &&
|
||||
(applicationStatus === 'archive_pending' || applicationStatus === 'failed'));
|
||||
if (isCommitted) {
|
||||
await this._discardStartupOps(startupOpIds);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Rejects the auto-generated startup example-task ops so they are NOT uploaded
|
||||
* after a SYNC_IMPORT is accepted silently. They were already excluded from the
|
||||
* conflict gate's "meaningful work" check (see SyncImportConflictGateService); the
|
||||
* import replaces local state, so rejecting them keeps the op-log consistent with
|
||||
* the just-applied remote data instead of re-uploading throwaway onboarding tasks.
|
||||
* Rejects startup-only ops so they are NOT uploaded after an authoritative remote
|
||||
* state is accepted silently. They were already excluded from the conflict gate's
|
||||
* "meaningful work" check (see SyncImportConflictGateService); rejecting them keeps
|
||||
* the op-log consistent with the just-applied remote data.
|
||||
*
|
||||
* These ids always come from getUnsynced() (local pending ops, never remote ops),
|
||||
* so a remote `isExampleTask` flag can never reach this path.
|
||||
* so a remote startup marker can never reach this path.
|
||||
*/
|
||||
private async _discardExampleTaskOps(opIds: string[]): Promise<void> {
|
||||
private async _discardStartupOps(opIds: string[]): Promise<void> {
|
||||
if (opIds.length > 0) {
|
||||
await this.opLogStore.markRejected(opIds);
|
||||
}
|
||||
|
|
@ -1260,8 +1343,24 @@ export class OperationLogSyncService {
|
|||
|
||||
const migratedRemoteOps = this._preflightRemoteOperations(result.newOps);
|
||||
|
||||
const currentSyncConfig = await firstValueFrom(this.store.select(selectSyncConfig));
|
||||
const localOnlySyncSettings: LocalOnlySyncSettings = {
|
||||
isEnabled: currentSyncConfig.isEnabled,
|
||||
isEncryptionEnabled: currentSyncConfig.isEncryptionEnabled,
|
||||
syncProvider: currentSyncConfig.syncProvider,
|
||||
syncInterval: currentSyncConfig.syncInterval,
|
||||
isManualSyncOnly: currentSyncConfig.isManualSyncOnly,
|
||||
};
|
||||
|
||||
let snapshotState = result.snapshotState as Record<string, unknown> | undefined;
|
||||
if (hasSnapshotState && snapshotState) {
|
||||
// File providers intentionally omit device-local schedule fields and null
|
||||
// the provider on the wire. Restore this device's values before schema
|
||||
// validation so a valid transport snapshot is locally replayable.
|
||||
snapshotState = applyLocalOnlySyncSettingsToAppData(
|
||||
stripLocalOnlySyncSettingsFromAppData(snapshotState),
|
||||
localOnlySyncSettings,
|
||||
) as Record<string, unknown>;
|
||||
const validation = await this.validateStateService.validateAndRepair(snapshotState);
|
||||
if (!validation.isValid) {
|
||||
throw new Error(
|
||||
|
|
@ -1298,14 +1397,6 @@ export class OperationLogSyncService {
|
|||
baselineGlobalConfig['sync'] && typeof baselineGlobalConfig['sync'] === 'object'
|
||||
? (baselineGlobalConfig['sync'] as Record<string, unknown>)
|
||||
: {};
|
||||
const currentSyncConfig = await firstValueFrom(this.store.select(selectSyncConfig));
|
||||
const localOnlySyncSettings: LocalOnlySyncSettings = {
|
||||
isEnabled: currentSyncConfig.isEnabled,
|
||||
isEncryptionEnabled: currentSyncConfig.isEncryptionEnabled,
|
||||
syncProvider: currentSyncConfig.syncProvider,
|
||||
syncInterval: currentSyncConfig.syncInterval,
|
||||
isManualSyncOnly: currentSyncConfig.isManualSyncOnly,
|
||||
};
|
||||
// getDefaultMainModelData intentionally excludes globalConfig. Add a
|
||||
// default config shell before applying the canonical device-local fields
|
||||
// so an interrupted rebuild can hydrate enough configuration to sync again.
|
||||
|
|
|
|||
|
|
@ -611,6 +611,40 @@ describe('RemoteOpsProcessingService', () => {
|
|||
expect(result.blockedByIncompatibleOp).toBe(true);
|
||||
});
|
||||
|
||||
it('should report a committed full-state prefix before a blocked suffix', async () => {
|
||||
const repair: Operation = {
|
||||
id: 'repair-prefix',
|
||||
opType: OpType.Repair,
|
||||
actionType: ActionType.REPAIR_AUTO,
|
||||
entityType: 'ALL',
|
||||
payload: {},
|
||||
clientId: 'client-1',
|
||||
vectorClock: { client1: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 1,
|
||||
};
|
||||
const futureOp: Operation = {
|
||||
id: 'future-suffix',
|
||||
opType: OpType.Update,
|
||||
actionType: ActionType.TASK_SHARED_UPDATE,
|
||||
entityType: 'TASK',
|
||||
entityId: 'task-1',
|
||||
payload: {},
|
||||
clientId: 'client-2',
|
||||
vectorClock: { client2: 1 },
|
||||
timestamp: Date.now(),
|
||||
schemaVersion: 2,
|
||||
};
|
||||
opLogStoreSpy.hasOp.and.resolveTo(false);
|
||||
opLogStoreSpy.append.and.resolveTo(1);
|
||||
|
||||
const result = await service.processRemoteOps([repair, futureOp]);
|
||||
|
||||
expect(result.blockedByIncompatibleOp).toBeTrue();
|
||||
expect(result.committedFullStateOpIds).toEqual([repair.id]);
|
||||
expect(operationApplierServiceSpy.applyOperations).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should show error snackbar and abort if version is below minimum supported', async () => {
|
||||
const remoteOps: Operation[] = [
|
||||
{ id: 'op1', schemaVersion: MIN_SUPPORTED_SCHEMA_VERSION - 1 } as Operation,
|
||||
|
|
|
|||
|
|
@ -7,9 +7,9 @@ import {
|
|||
ConflictResult,
|
||||
EntityConflict,
|
||||
extractFullStateFromPayload,
|
||||
FULL_STATE_OP_TYPES,
|
||||
isWrappedFullStatePayload,
|
||||
Operation,
|
||||
OpType,
|
||||
VectorClock,
|
||||
} from '../core/operation.types';
|
||||
import { OpLog } from '../../core/log';
|
||||
|
|
@ -105,6 +105,12 @@ export class RemoteOpsProcessingService {
|
|||
filteringImport?: Operation;
|
||||
isLocalUnsyncedImport: boolean;
|
||||
blockedByIncompatibleOp: boolean;
|
||||
/**
|
||||
* Full-state operations whose application completed (or was already
|
||||
* deduplicated as applied) before this result was returned. Populated even
|
||||
* when a later incompatible op blocks the remaining batch suffix.
|
||||
*/
|
||||
committedFullStateOpIds?: string[];
|
||||
}> {
|
||||
// Validation failure surfaces via the SyncSessionValidationService latch
|
||||
// (#7330). `validateAfterSync` and the conflict-resolution validation path
|
||||
|
|
@ -248,18 +254,16 @@ export class RemoteOpsProcessingService {
|
|||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// STEP 3: Check for full-state operations (SYNC_IMPORT / BACKUP_IMPORT)
|
||||
// STEP 3: Check for full-state operations
|
||||
// These replace the entire state, so conflict detection doesn't apply.
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
const hasFullStateOp = validOps.some(
|
||||
(op) => op.opType === OpType.SyncImport || op.opType === OpType.BackupImport,
|
||||
);
|
||||
const hasFullStateOp = validOps.some((op) => FULL_STATE_OP_TYPES.has(op.opType));
|
||||
|
||||
if (hasFullStateOp) {
|
||||
OpLog.normal(
|
||||
'RemoteOpsProcessingService: Full-state operation detected, skipping conflict detection.',
|
||||
);
|
||||
await this.applyNonConflictingOps(validOps);
|
||||
const committedFullStateOpIds = await this.applyNonConflictingOps(validOps);
|
||||
|
||||
// Clean Slate Semantics: SYNC_IMPORT/BACKUP_IMPORT replaces entire state.
|
||||
// Local synced ops are NOT replayed - the import is an explicit user action
|
||||
|
|
@ -272,6 +276,7 @@ export class RemoteOpsProcessingService {
|
|||
filteredOpCount: 0,
|
||||
isLocalUnsyncedImport: false,
|
||||
blockedByIncompatibleOp,
|
||||
committedFullStateOpIds,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -434,7 +439,7 @@ export class RemoteOpsProcessingService {
|
|||
async applyNonConflictingOps(
|
||||
ops: Operation[],
|
||||
callerHoldsLock: boolean = false,
|
||||
): Promise<void> {
|
||||
): Promise<string[]> {
|
||||
const locallyReplayableOps =
|
||||
await this._withLocalOnlySyncSettingsForFullStateOps(ops);
|
||||
|
||||
|
|
@ -447,6 +452,7 @@ export class RemoteOpsProcessingService {
|
|||
// to leak into the next sync window with stale clocks. (#7700)
|
||||
let didApplyRemoteOps = false;
|
||||
let primaryIncompleteError: IncompleteRemoteOperationsError | undefined;
|
||||
let committedFullStateOpIds: string[] = [];
|
||||
try {
|
||||
// Core owns the generic crash-safety ordering. Angular diagnostics,
|
||||
// validation, and user notifications stay in this service.
|
||||
|
|
@ -464,6 +470,9 @@ export class RemoteOpsProcessingService {
|
|||
});
|
||||
|
||||
didApplyRemoteOps = result.appendedOps.length > 0;
|
||||
committedFullStateOpIds = result.appendedOps
|
||||
.filter((op) => FULL_STATE_OP_TYPES.has(op.opType))
|
||||
.map((op) => op.id);
|
||||
|
||||
if (result.skippedCount > 0) {
|
||||
OpLog.verbose(
|
||||
|
|
@ -526,6 +535,7 @@ export class RemoteOpsProcessingService {
|
|||
}
|
||||
}
|
||||
}
|
||||
return committedFullStateOpIds;
|
||||
}
|
||||
|
||||
private async _withLocalOnlySyncSettingsForFullStateOps(
|
||||
|
|
@ -569,11 +579,7 @@ export class RemoteOpsProcessingService {
|
|||
}
|
||||
|
||||
private _isFullStateOperation(op: Operation): boolean {
|
||||
return (
|
||||
op.opType === OpType.SyncImport ||
|
||||
op.opType === OpType.BackupImport ||
|
||||
op.opType === OpType.Repair
|
||||
);
|
||||
return FULL_STATE_OP_TYPES.has(op.opType);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -118,13 +118,38 @@ describe('SyncImportConflictGateService', () => {
|
|||
expect(result.dialogData).toBeDefined();
|
||||
});
|
||||
|
||||
it('should protect a pending sync config change on a never-synced client', async () => {
|
||||
it('should ignore the sync setup write on a never-synced client', async () => {
|
||||
opLogStoreSpy.hasSyncedOps.and.resolveTo(false);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([createPendingConfigEntry()]);
|
||||
|
||||
const result = await service.checkIncomingFullStateConflict([createOperation()]);
|
||||
|
||||
expect(result.hasMeaningfulPending).toBeFalse();
|
||||
expect(result.discardablePendingOpIds).toEqual(['local-config-update']);
|
||||
expect(result.dialogData).toBeUndefined();
|
||||
});
|
||||
|
||||
it('should protect non-update operations targeting the sync config coordinates', async () => {
|
||||
opLogStoreSpy.hasSyncedOps.and.resolveTo(false);
|
||||
opLogStoreSpy.getUnsynced.and.resolveTo([
|
||||
createEntry(
|
||||
createOperation({
|
||||
id: 'local-config-delete',
|
||||
actionType: ActionType.GLOBAL_CONFIG_UPDATE_SECTION,
|
||||
opType: OpType.Delete,
|
||||
entityType: 'GLOBAL_CONFIG',
|
||||
entityId: 'sync',
|
||||
payload: { sectionKey: 'sync' },
|
||||
clientId: 'client-A',
|
||||
vectorClock: { clientA: 1 },
|
||||
}),
|
||||
),
|
||||
]);
|
||||
|
||||
const result = await service.checkIncomingFullStateConflict([createOperation()]);
|
||||
|
||||
expect(result.hasMeaningfulPending).toBeTrue();
|
||||
expect(result.discardablePendingOpIds).toEqual([]);
|
||||
expect(result.dialogData).toBeDefined();
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import { inject, Injectable } from '@angular/core';
|
||||
import { OperationLogStoreService } from '../persistence/operation-log-store.service';
|
||||
import {
|
||||
ActionType,
|
||||
FULL_STATE_OP_TYPES,
|
||||
Operation,
|
||||
OperationLogEntry,
|
||||
|
|
@ -10,6 +11,21 @@ import { OperationWriteFlushService } from './operation-write-flush.service';
|
|||
import { SyncImportConflictData } from './dialog-sync-import-conflict/dialog-sync-import-conflict.component';
|
||||
import { isExampleTaskCreateOp } from '../validation/is-example-task-op.util';
|
||||
|
||||
/**
|
||||
* Enabling/configuring sync on a never-synced client necessarily writes the
|
||||
* sync config section before the first download. That setup-only operation is
|
||||
* not local user data divergence. Other GLOBAL_CONFIG sections remain protected.
|
||||
*/
|
||||
const isInitialSyncSetupOp = (entry: OperationLogEntry): boolean =>
|
||||
entry.op.actionType === ActionType.GLOBAL_CONFIG_UPDATE_SECTION &&
|
||||
entry.op.opType === OpType.Update &&
|
||||
entry.op.entityType === 'GLOBAL_CONFIG' &&
|
||||
entry.op.entityId === 'sync';
|
||||
|
||||
interface PendingOpClassificationOptions {
|
||||
hasCompletedInitialSync: boolean;
|
||||
}
|
||||
|
||||
export interface IncomingFullStateConflictGateResult {
|
||||
fullStateOp?: Operation;
|
||||
pendingOps: OperationLogEntry[];
|
||||
|
|
@ -34,14 +50,20 @@ export class SyncImportConflictGateService {
|
|||
private writeFlushService = inject(OperationWriteFlushService);
|
||||
|
||||
/**
|
||||
* Every pending op is user work unless it is an onboarding example-task create.
|
||||
* Every pending op is user work unless it is an onboarding example-task create,
|
||||
* or the sync-section setup write on a client that has never completed sync.
|
||||
* Full-state ops are always meaningful because applying a newer full-state op
|
||||
* can invalidate their local import/repair semantics.
|
||||
*
|
||||
* The lifecycle default is deliberately conservative: a caller that does not
|
||||
* know whether initial sync completed must protect startup-entity changes.
|
||||
*/
|
||||
hasMeaningfulPendingOps(ops: OperationLogEntry[]): boolean {
|
||||
hasMeaningfulPendingOps(
|
||||
ops: OperationLogEntry[],
|
||||
options: PendingOpClassificationOptions = {
|
||||
hasCompletedInitialSync: true,
|
||||
},
|
||||
): boolean {
|
||||
return ops.some((entry) => {
|
||||
if (FULL_STATE_OP_TYPES.has(entry.op.opType as OpType)) {
|
||||
return true;
|
||||
|
|
@ -49,10 +71,26 @@ export class SyncImportConflictGateService {
|
|||
if (isExampleTaskCreateOp(entry)) {
|
||||
return false;
|
||||
}
|
||||
if (isInitialSyncSetupOp(entry)) {
|
||||
return options.hasCompletedInitialSync;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
getDiscardablePendingOpIds(
|
||||
ops: OperationLogEntry[],
|
||||
options: PendingOpClassificationOptions,
|
||||
): string[] {
|
||||
return ops
|
||||
.filter(
|
||||
(entry) =>
|
||||
isExampleTaskCreateOp(entry) ||
|
||||
(!options.hasCompletedInitialSync && isInitialSyncSetupOp(entry)),
|
||||
)
|
||||
.map((entry) => entry.op.id);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param options.preCapturedPendingOps - Exact pending ops selected by the upload
|
||||
* round. The piggyback path unions this snapshot with a live read so it
|
||||
|
|
@ -89,14 +127,6 @@ export class SyncImportConflictGateService {
|
|||
? this._mergePendingOps(options.preCapturedPendingOps, livePendingOps)
|
||||
: livePendingOps;
|
||||
|
||||
// Example-task ops that the caller may reject when it accepts the import silently.
|
||||
// These must come from a LIVE read: with a pre-captured snapshot, example ops
|
||||
// accepted earlier in the same upload round are already marked synced and must
|
||||
// not be re-marked rejected by the caller.
|
||||
const discardablePendingOpIds = livePendingOps
|
||||
.filter(isExampleTaskCreateOp)
|
||||
.map((entry) => entry.op.id);
|
||||
|
||||
// Preserve the cheap example-task-only path. If nothing could be meaningful
|
||||
// even for a synced client, there is no reason to read sync history.
|
||||
const canContainMeaningfulPending = this.hasMeaningfulPendingOps(pendingOps);
|
||||
|
|
@ -105,13 +135,28 @@ export class SyncImportConflictGateService {
|
|||
fullStateOp,
|
||||
pendingOps,
|
||||
hasMeaningfulPending: false,
|
||||
discardablePendingOpIds,
|
||||
discardablePendingOpIds: this.getDiscardablePendingOpIds(livePendingOps, {
|
||||
hasCompletedInitialSync: true,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
const isNeverSynced =
|
||||
options.isNeverSynced ?? !(await this.opLogStore.hasSyncedOps());
|
||||
const hasMeaningfulPending = this.hasMeaningfulPendingOps(pendingOps);
|
||||
const classificationOptions = {
|
||||
hasCompletedInitialSync: !isNeverSynced,
|
||||
};
|
||||
const hasMeaningfulPending = this.hasMeaningfulPendingOps(
|
||||
pendingOps,
|
||||
classificationOptions,
|
||||
);
|
||||
// Only live pending ops may be rejected. A pre-captured upload snapshot can
|
||||
// include ops already acknowledged by the server, which must not be rewritten
|
||||
// as rejected locally.
|
||||
const discardablePendingOpIds = this.getDiscardablePendingOpIds(
|
||||
livePendingOps,
|
||||
classificationOptions,
|
||||
);
|
||||
|
||||
const result = {
|
||||
fullStateOp,
|
||||
|
|
|
|||
|
|
@ -36,13 +36,13 @@ describe('Example-task SYNC_IMPORT gate (integration)', () => {
|
|||
},
|
||||
});
|
||||
|
||||
const configOp = (): Operation =>
|
||||
const configOp = (sectionKey = 'sync'): Operation =>
|
||||
local.createOperation({
|
||||
actionType: ActionType.GLOBAL_CONFIG_UPDATE_SECTION,
|
||||
opType: OpType.Update,
|
||||
entityType: 'GLOBAL_CONFIG',
|
||||
entityId: 'sync',
|
||||
payload: { sectionKey: 'sync' },
|
||||
entityId: sectionKey,
|
||||
payload: { sectionKey },
|
||||
});
|
||||
|
||||
const incomingSyncImport = (): Operation =>
|
||||
|
|
@ -89,24 +89,20 @@ describe('Example-task SYNC_IMPORT gate (integration)', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('treats a pending config change as meaningful user work (widened gate) and shows the dialog', async () => {
|
||||
// The gate deliberately has NO entity-wide exemptions beyond example-task
|
||||
// creates: GLOBAL_CONFIG carries synced preferences, so silently discarding
|
||||
// a pending config op on an incoming SYNC_IMPORT would lose user work.
|
||||
it('treats the never-synced provider setup write as discardable startup work', async () => {
|
||||
const example = exampleTaskOp('example-task-1');
|
||||
await storeService.append(configOp(), 'local');
|
||||
const config = configOp();
|
||||
await storeService.append(config, 'local');
|
||||
await storeService.append(example, 'local');
|
||||
|
||||
const result = await gate.checkIncomingFullStateConflict([incomingSyncImport()]);
|
||||
|
||||
expect(result.hasMeaningfulPending).toBeTrue();
|
||||
expect(result.dialogData).toBeDefined();
|
||||
// Example ops stay listed so the caller can discard them if the user
|
||||
// accepts the import.
|
||||
expect(result.discardablePendingOpIds).toEqual([example.id]);
|
||||
expect(result.hasMeaningfulPending).toBeFalse();
|
||||
expect(result.dialogData).toBeUndefined();
|
||||
expect(result.discardablePendingOpIds.sort()).toEqual([config.id, example.id].sort());
|
||||
});
|
||||
|
||||
it('actually excludes example-task ops from getUnsynced after markRejected (so they are not uploaded)', async () => {
|
||||
it('actually excludes discardable startup ops from getUnsynced after markRejected', async () => {
|
||||
const config = configOp();
|
||||
const example = exampleTaskOp('example-task-1');
|
||||
await storeService.append(config, 'local');
|
||||
|
|
@ -116,10 +112,21 @@ describe('Example-task SYNC_IMPORT gate (integration)', () => {
|
|||
await storeService.markRejected(result.discardablePendingOpIds);
|
||||
|
||||
const remaining = (await storeService.getUnsynced()).map((e) => e.op.id);
|
||||
expect(remaining).toContain(config.id);
|
||||
expect(remaining).not.toContain(config.id);
|
||||
expect(remaining).not.toContain(example.id);
|
||||
});
|
||||
|
||||
it('keeps another config section meaningful on a never-synced client', async () => {
|
||||
const config = configOp('productivityHacks');
|
||||
await storeService.append(config, 'local');
|
||||
|
||||
const result = await gate.checkIncomingFullStateConflict([incomingSyncImport()]);
|
||||
|
||||
expect(result.hasMeaningfulPending).toBeTrue();
|
||||
expect(result.dialogData).toBeDefined();
|
||||
expect(result.discardablePendingOpIds).toEqual([]);
|
||||
});
|
||||
|
||||
it('shows the dialog (and still lists the example id) when real user work is also pending', async () => {
|
||||
const example = exampleTaskOp('example-task-1');
|
||||
const realTaskUpdate = local.createOperation({
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue