diff --git a/src/app/op-log/persistence/operation-log-store.service.spec.ts b/src/app/op-log/persistence/operation-log-store.service.spec.ts index 422d7e8a97..ee4b544471 100644 --- a/src/app/op-log/persistence/operation-log-store.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-store.service.spec.ts @@ -2404,6 +2404,33 @@ describe('OperationLogStoreService', () => { expect(await service.isRawRebuildIncomplete()).toBe(false); }); + it('durably carries post-crash local ops in the rebuild marker', async () => { + const preservedLocalOp = createTestOperation({ + id: '01900000-0000-7000-8000-000000000091', + entityId: 'edited-after-crash', + clientId: 'localClient', + vectorClock: { localClient: 2, remote: 1 }, + }); + + await service.runRemoteStateReplacement({ + baselineState: { task: { ids: [], entities: {} } }, + vectorClock: { remote: 1 }, + schemaVersion: 4, + snapshotEntityKeys: [], + archiveYoung: createArchive('young'), + archiveOld: createArchive('old'), + preservedLocalOps: [preservedLocalOp], + }); + + expect(await service.getOpsAfterSeq(0)).toEqual([]); + expect(await service.loadRawRebuildIncomplete()).toEqual( + jasmine.objectContaining({ + incomplete: true, + preservedLocalOps: [preservedLocalOp], + }), + ); + }); + it('rolls back every store if one archive write fails', async () => { const priorOp = createTestOperation({ entityId: 'prior-task' }); const priorState = { sentinel: 'prior-state' }; diff --git a/src/app/op-log/persistence/operation-log-store.service.ts b/src/app/op-log/persistence/operation-log-store.service.ts index 1b387219ba..5bb985048c 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -80,6 +80,12 @@ interface StateCacheEntry { snapshotEntityKeys?: string[]; } +export interface RawRebuildIncompleteEntry { + incomplete: true; + startedAt: number; + preservedLocalOps: Operation[]; +} + /** * Stored operation log entry that can hold either compact or full operation format. * Used internally for backwards compatibility with existing data. @@ -1556,6 +1562,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); @@ -1584,7 +1591,11 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + return (await this.loadRawRebuildIncomplete()) !== null; + } + + /** + * Loads the durable resume marker, including local operations created after + * an interrupted replacement. Older markers did not contain the operation + * array, so they normalize to an empty list. + */ + async loadRawRebuildIncomplete(): Promise { await this._ensureInit(); - const entry = await this._adapter.get<{ incomplete?: boolean }>( + const entry = await this._adapter.get>( STORE_NAMES.META, RAW_REBUILD_INCOMPLETE_META_KEY, ); - return entry?.incomplete === true; + if (entry?.incomplete !== true) { + return null; + } + return { + incomplete: true, + startedAt: typeof entry.startedAt === 'number' ? entry.startedAt : 0, + preservedLocalOps: Array.isArray(entry.preservedLocalOps) + ? entry.preservedLocalOps + : [], + }; } /** Marks the USE_REMOTE raw rebuild as fully completed. */ diff --git a/src/app/op-log/sync/operation-log-sync.service.spec.ts b/src/app/op-log/sync/operation-log-sync.service.spec.ts index 05e0cbdd62..1189c3f3a4 100644 --- a/src/app/op-log/sync/operation-log-sync.service.spec.ts +++ b/src/app/op-log/sync/operation-log-sync.service.spec.ts @@ -57,6 +57,7 @@ describe('OperationLogSyncService', () => { let schemaMigrationServiceSpy: jasmine.SpyObj; let validateStateServiceSpy: jasmine.SpyObj; let lockServiceSpy: jasmine.SpyObj; + let operationApplierSpy: jasmine.SpyObj; beforeEach(() => { snackServiceSpy = jasmine.createSpyObj('SnackService', [ @@ -78,10 +79,12 @@ describe('OperationLogSyncService', () => { 'hasSyncedOps', 'runRemoteStateReplacement', 'isRawRebuildIncomplete', + 'loadRawRebuildIncomplete', 'clearRawRebuildIncomplete', 'loadImportBackup', ]); opLogStoreSpy.hasSyncedOps.and.resolveTo(true); + opLogStoreSpy.getUnsynced.and.resolveTo([]); opLogStoreSpy.markSynced.and.resolveTo(); opLogStoreSpy.setVectorClock.and.resolveTo(); opLogStoreSpy.clearFullStateOps.and.resolveTo(); @@ -93,6 +96,7 @@ describe('OperationLogSyncService', () => { }); opLogStoreSpy.runRemoteStateReplacement.and.resolveTo(); opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(false); + opLogStoreSpy.loadRawRebuildIncomplete.and.resolveTo(null); opLogStoreSpy.clearRawRebuildIncomplete.and.resolveTo(); opLogStoreSpy.loadImportBackup.and.resolveTo(null); @@ -180,6 +184,10 @@ describe('OperationLogSyncService', () => { ['showConflictDialog'], ); syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL'); + operationApplierSpy = jasmine.createSpyObj('OperationApplierService', [ + 'applyOperations', + ]); + operationApplierSpy.applyOperations.and.resolveTo({ appliedOps: [] }); TestBed.configureTestingModule({ providers: [ @@ -199,7 +207,7 @@ describe('OperationLogSyncService', () => { }, { provide: OperationApplierService, - useValue: jasmine.createSpyObj('OperationApplierService', ['applyOperations']), + useValue: operationApplierSpy, }, { provide: ConflictResolutionService, @@ -2739,6 +2747,70 @@ describe('OperationLogSyncService', () => { ); }); + it('should preserve and replay local edits made after an interrupted rebuild', async () => { + const previouslyPreserved = makeRemoteOp('local-before-second-crash'); + previouslyPreserved.clientId = 'local'; + previouslyPreserved.vectorClock = { local: 2, remote: 1 }; + const liveLocalEdit = makeRemoteOp('local-after-restart'); + liveLocalEdit.clientId = 'local'; + liveLocalEdit.vectorClock = { local: 3, remote: 1 }; + const liveEntry = { + seq: 2, + op: liveLocalEdit, + source: 'local' as const, + appliedAt: Date.now(), + }; + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp('remote-op')], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 4, + }); + opLogStoreSpy.loadImportBackup.and.resolveTo({ state: {}, savedAt: 12345 }); + opLogStoreSpy.loadRawRebuildIncomplete.and.resolveTo({ + incomplete: true, + startedAt: 1, + preservedLocalOps: [previouslyPreserved], + }); + opLogStoreSpy.getUnsynced.and.resolveTo([liveEntry]); + opLogStoreSpy.appendBatchSkipDuplicates.and.callFake(async (ops, source) => ({ + seqs: ops.map((_, index) => index + 10), + writtenOps: source === 'local' ? ops : [], + skippedCount: 0, + })); + operationApplierSpy.applyOperations.and.callFake(async (ops) => ({ + appliedOps: ops, + })); + opLogStoreSpy.getVectorClock.and.resolveTo({ remote: 4 }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await service.forceDownloadRemoteState(mockProvider, { isCrashResume: true }); + + const preservedLocalOps = [previouslyPreserved, liveLocalEdit]; + expect( + opLogStoreSpy.runRemoteStateReplacement.calls.mostRecent().args[0] + .preservedLocalOps, + ).toEqual(preservedLocalOps); + expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith( + preservedLocalOps, + 'local', + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith( + preservedLocalOps, + jasmine.objectContaining({ isLocalHydration: false }), + ); + expect(opLogStoreSpy.setVectorClock).toHaveBeenCalledWith({ + remote: 4, + local: 3, + }); + expect(opLogStoreSpy.clearRawRebuildIncomplete).toHaveBeenCalled(); + }); + it('should offer to restore the previous data after replacing (#8107)', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [makeRemoteOp()], diff --git a/src/app/op-log/sync/operation-log-sync.service.ts b/src/app/op-log/sync/operation-log-sync.service.ts index ecc341401b..0aff4ddf64 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -60,6 +60,7 @@ import { LocalOnlySyncSettings, } 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'; /** * Orchestrates synchronization of the Operation Log with remote storage. @@ -149,6 +150,7 @@ export class OperationLogSyncService { private syncLocalStateService = inject(SyncLocalStateService); private syncImportConflictCoordinator = inject(SyncImportConflictCoordinatorService); private providerManager = inject(SyncProviderManager); + private operationApplier = inject(OperationApplierService); /** * Checks if this client is "wholly fresh" - meaning it has never synced before @@ -1333,6 +1335,7 @@ export class OperationLogSyncService { let capturedBackupSavedAt: number | undefined; let replacementCommitted = false; let backupSavedAt: number | undefined; + let preservedLocalOps: Operation[] = []; try { // flushThenRunExclusive drains the capture pipeline BEFORE acquiring the // op-log lock (flushPendingWrites re-acquires the same non-reentrant lock, @@ -1345,6 +1348,14 @@ export class OperationLogSyncService { // Keep the first attempt's pre-replace backup (see option JSDoc). savedAt = (await this.opLogStore.loadImportBackup())?.savedAt; capturedBackupSavedAt = savedAt; + const marker = await this.opLogStore.loadRawRebuildIncomplete(); + const liveLocalOps = (await this.opLogStore.getUnsynced()).map( + (entry) => entry.op, + ); + preservedLocalOps = this._mergeOperationsById( + marker?.preservedLocalOps ?? [], + liveLocalOps, + ); } else { try { savedAt = await this.backupService.captureImportBackup(); @@ -1375,6 +1386,7 @@ export class OperationLogSyncService { ), archiveYoung, archiveOld, + preservedLocalOps, }); replacementCommitted = true; @@ -1417,6 +1429,8 @@ export class OperationLogSyncService { ); } + await this._restorePreservedLocalOps(preservedLocalOps); + // Update lastServerSeq after hydration if (result.latestServerSeq !== undefined) { await syncProvider.setLastServerSeq(result.latestServerSeq); @@ -1467,6 +1481,8 @@ export class OperationLogSyncService { ); } + await this._restorePreservedLocalOps(preservedLocalOps); + // Update lastServerSeq if (result.latestServerSeq !== undefined) { await syncProvider.setLastServerSeq(result.latestServerSeq); @@ -1494,6 +1510,57 @@ export class OperationLogSyncService { } } + private _mergeOperationsById(...operationGroups: Operation[][]): Operation[] { + const merged: Operation[] = []; + const seenIds = new Set(); + for (const operations of operationGroups) { + for (const op of operations) { + if (!seenIds.has(op.id)) { + seenIds.add(op.id); + merged.push(op); + } + } + } + return merged; + } + + /** + * Replays edits captured after an interrupted rebuild on top of the complete + * authoritative history. They stay local/unsynced so the next upload carries + * the user's post-crash intent to the server. + */ + private async _restorePreservedLocalOps(operations: Operation[]): Promise { + if (operations.length === 0) { + return; + } + + const { writtenOps } = await this.opLogStore.appendBatchSkipDuplicates( + operations, + 'local', + ); + if (writtenOps.length === 0) { + return; + } + + const applyResult = await this.operationApplier.applyOperations(writtenOps, { + // The authoritative replacement also overwrote archive IndexedDB stores, + // so archive-affecting post-crash edits must replay their side effects. + // The entries themselves remain source=local and unsynced in the op-log. + isLocalHydration: false, + }); + if (applyResult.failedOp || applyResult.appliedOps.length !== writtenOps.length) { + throw new Error( + 'USE_REMOTE incomplete: post-crash local operations could not be restored.', + ); + } + + let restoredClock = (await this.opLogStore.getVectorClock()) ?? {}; + for (const op of writtenOps) { + restoredClock = mergeVectorClocks(restoredClock, op.vectorClock); + } + await this.opLogStore.setVectorClock(restoredClock); + } + private _preflightRemoteOperations(remoteOps: Operation[]): Operation[] { for (const op of remoteOps) { let version: number;