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 e34fa91ddf..88c62a442a 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 @@ -1560,6 +1560,167 @@ describe('OperationLogStoreService', () => { const storedOps = await service.getOpsAfterSeq(0); expect(storedOps.length).toBe(1); }); + + it('should append snapshot-included ops and atomically advance the state-cache frontier', async () => { + const existingOp = createTestOperation({ id: 'snapshot-op-existing' }); + const newOp = createTestOperation({ id: 'snapshot-op-new' }); + await service.append(existingOp, 'remote'); + await service.saveStateCache({ + state: { task: { ids: ['task1'] } }, + lastAppliedOpSeq: 1, + vectorClock: { testClient: 1 }, + compactedAt: 1, + }); + + const result = await service.appendSnapshotIncludedOps([existingOp, newOp]); + + expect(result.writtenOps).toEqual([newOp]); + expect(result.skippedCount).toBe(1); + expect((await service.loadStateCache())?.lastAppliedOpSeq).toBe(2); + }); + + it('should not append snapshot-included ops without an existing state cache', async () => { + const snapshotOp = createTestOperation({ id: 'snapshot-op-without-cache' }); + + await expectAsync( + service.appendSnapshotIncludedOps([snapshotOp]), + ).toBeRejectedWithError( + 'Cannot append snapshot-included operations without an existing state cache', + ); + + expect(await service.getOpsAfterSeq(0)).toEqual([]); + expect(await service.loadStateCache()).toBeNull(); + }); + + it('should reject a snapshot append when the state-cache frontier is behind the log tail', async () => { + const existingOp = createTestOperation({ id: 'existing-unmaterialized-op' }); + const snapshotOp = createTestOperation({ id: 'snapshot-op-after-gap' }); + await service.append(existingOp, 'remote'); + await service.saveStateCache({ + state: { task: { ids: [] } }, + lastAppliedOpSeq: 0, + vectorClock: {}, + compactedAt: 1, + }); + + await expectAsync( + service.appendSnapshotIncludedOps([snapshotOp]), + ).toBeRejectedWithError( + 'Cannot append snapshot-included operations when the state-cache frontier does not match the operation-log tail', + ); + + expect((await service.getOpsAfterSeq(0)).map((entry) => entry.op.id)).toEqual([ + existingOp.id, + ]); + expect((await service.loadStateCache())?.lastAppliedOpSeq).toBe(0); + }); + + it('should atomically commit file snapshot ops, cache, clock, and archives', async () => { + const existingOp = createTestOperation({ id: 'file-snapshot-existing' }); + const includedOp = createTestOperation({ id: 'file-snapshot-new' }); + const archiveYoung = { + task: { ids: [], entities: {} }, + timeTracking: { project: {}, tag: {} }, + } as unknown as ArchiveModel; + const archiveOld = { + task: { ids: [], entities: {} }, + timeTracking: { project: {}, tag: {} }, + } as unknown as ArchiveModel; + await service.append(existingOp, 'remote'); + + const result = await service.commitFileSnapshotBaseline({ + state: { task: { ids: ['remote-task'] } }, + lastAppliedOpSeq: 1, + vectorClock: { remote: 7 }, + compactedAt: 123, + snapshotIncludedOps: [existingOp, includedOp], + archiveYoung, + archiveOld, + }); + + expect(result).toEqual({ + seqs: [2], + writtenOps: [includedOp], + skippedCount: 1, + }); + expect((await service.loadStateCache())?.lastAppliedOpSeq).toBe(2); + expect(await service.getVectorClock()).toEqual({ remote: 7 }); + const db = ( + service as unknown as { + db: IDBPDatabase; + } + ).db; + expect((await db.get(STORE_NAMES.ARCHIVE_YOUNG, SINGLETON_KEY)).data).toEqual( + archiveYoung, + ); + expect((await db.get(STORE_NAMES.ARCHIVE_OLD, SINGLETON_KEY)).data).toEqual( + archiveOld, + ); + }); + + it('should roll back the whole file baseline when its vector-clock write fails', async () => { + const priorOp = createTestOperation({ id: 'file-snapshot-prior-op' }); + const priorState = { sentinel: 'prior-state' }; + await service.append(priorOp, 'remote'); + await service.saveStateCache({ + state: priorState, + lastAppliedOpSeq: 1, + vectorClock: { testClient: 1 }, + compactedAt: 1, + }); + await service.setVectorClock({ testClient: 1 }); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + const originalTransaction = adapter.transaction.bind(adapter); + spyOn(adapter, 'transaction').and.callFake(async (stores, mode, callback) => + originalTransaction(stores, mode, async (tx) => { + const failingTx = new Proxy(tx, { + get: (target, property): unknown => { + if (property === 'put') { + return async ( + storeName: Parameters[0], + ...args: unknown[] + ): Promise => { + if (storeName === STORE_NAMES.VECTOR_CLOCK) { + throw new Error('injected vector-clock write failure'); + } + return (target.put as (...putArgs: unknown[]) => Promise).call( + target, + storeName, + ...args, + ); + }; + } + const value = Reflect.get(target, property); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + return callback(failingTx); + }), + ); + + await expectAsync( + service.commitFileSnapshotBaseline({ + state: { sentinel: 'new-state' }, + lastAppliedOpSeq: 1, + vectorClock: { remote: 2 }, + compactedAt: 2, + snapshotIncludedOps: [ + createTestOperation({ id: 'file-snapshot-rolled-back-op' }), + ], + }), + ).toBeRejectedWithError('injected vector-clock write failure'); + + expect((await service.getOpsAfterSeq(0)).map(({ op }) => op.id)).toEqual([ + priorOp.id, + ]); + expect((await service.loadStateCache())?.state).toEqual(priorState); + expect(await service.getVectorClock()).toEqual({ testClient: 1 }); + }); }); describe('appendMixedSourceBatchSkipDuplicates', () => { 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 29160dce5c..048dc686a9 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -783,6 +783,151 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + return this._appendBatchSkipDuplicates(ops, source, options, false); + } + + /** + * Records remote operations already materialized by the current state cache. + * + * The cache must be exactly at the pre-append operation-log tail. This keeps + * its contiguous replay frontier from skipping unrelated operations while + * still allowing the supplied snapshot operations to be appended and + * checkpointed atomically. + */ + async appendSnapshotIncludedOps( + ops: Operation[], + ): Promise<{ seqs: number[]; writtenOps: Operation[]; skippedCount: number }> { + return this._appendBatchSkipDuplicates(ops, 'remote', undefined, true); + } + + /** + * Atomically commits every durable part of a file-snapshot baseline. + * + * The downloaded state, its archives/vector clock, and the remote operations + * already represented by that state are one commit point. Keeping them in a + * single transaction means a failed write leaves the previous baseline fully + * intact, so local actions deferred during the download can safely be written + * against it instead of being stranded behind a partial snapshot. + */ + async commitFileSnapshotBaseline(opts: { + state: unknown; + lastAppliedOpSeq: number; + vectorClock: VectorClock; + compactedAt: number; + snapshotIncludedOps: readonly Operation[]; + archiveYoung?: ArchiveStoreEntry['data']; + archiveOld?: ArchiveStoreEntry['data']; + }): Promise<{ seqs: number[]; writtenOps: Operation[]; skippedCount: number }> { + await this._ensureInit(); + + const storeNames: OpLogStoreName[] = [ + STORE_NAMES.OPS, + STORE_NAMES.STATE_CACHE, + STORE_NAMES.VECTOR_CLOCK, + ]; + if (opts.snapshotIncludedOps.some((op) => isFullStateOpType(op.opType))) { + storeNames.push(STORE_NAMES.META); + } + if (opts.archiveYoung !== undefined) { + storeNames.push(STORE_NAMES.ARCHIVE_YOUNG); + } + if (opts.archiveOld !== undefined) { + storeNames.push(STORE_NAMES.ARCHIVE_OLD); + } + + try { + const result = await this._lockService.request(LOCK_NAMES.TASK_ARCHIVE, () => + this._adapter.transaction(storeNames, 'readwrite', async (tx) => { + let preAppendLastSeq = 0; + await tx.iterate( + STORE_NAMES.OPS, + { direction: 'prev' }, + (_value, key) => { + if (typeof key !== 'number') { + throw new Error('Operation sequence key is not numeric'); + } + preAppendLastSeq = key; + return 'stop'; + }, + ); + if (preAppendLastSeq !== opts.lastAppliedOpSeq) { + throw new Error( + 'Cannot commit a file snapshot after the operation-log tail changed', + ); + } + + const seqs: number[] = []; + const writtenOps: Operation[] = []; + let skippedCount = 0; + let snapshotFrontier = preAppendLastSeq; + for (const op of opts.snapshotIncludedOps) { + const existingKey = await tx.getKeyFromIndex( + STORE_NAMES.OPS, + OPS_INDEXES.BY_ID, + op.id, + ); + if (existingKey !== undefined) { + if (typeof existingKey !== 'number') { + throw new Error('Operation sequence key is not numeric'); + } + snapshotFrontier = Math.max(snapshotFrontier, existingKey); + skippedCount++; + continue; + } + + const entry = this._buildStoredEntry(op, 'remote'); + const seq = await tx.add(STORE_NAMES.OPS, entry); + await this._recordFullStateOpInTx(tx, entry.op, seq); + snapshotFrontier = Math.max(snapshotFrontier, seq); + seqs.push(seq); + writtenOps.push(op); + } + + await tx.put(STORE_NAMES.STATE_CACHE, { + id: SINGLETON_KEY, + state: opts.state, + lastAppliedOpSeq: snapshotFrontier, + vectorClock: opts.vectorClock, + compactedAt: opts.compactedAt, + } satisfies StateCacheEntry); + await tx.put( + STORE_NAMES.VECTOR_CLOCK, + { clock: opts.vectorClock, lastUpdate: opts.compactedAt }, + SINGLETON_KEY, + ); + if (opts.archiveYoung !== undefined) { + await tx.put(STORE_NAMES.ARCHIVE_YOUNG, { + id: SINGLETON_KEY, + data: opts.archiveYoung, + lastModified: opts.compactedAt, + } satisfies ArchiveStoreEntry); + } + if (opts.archiveOld !== undefined) { + await tx.put(STORE_NAMES.ARCHIVE_OLD, { + id: SINGLETON_KEY, + data: opts.archiveOld, + lastModified: opts.compactedAt, + } satisfies ArchiveStoreEntry); + } + + return { seqs, writtenOps, skippedCount }; + }), + ); + + this._vectorClockCache = { ...opts.vectorClock }; + this._invalidateAppliedAndUnsyncedCaches(); + return result; + } catch (e) { + this._handleAppendError(e); + } + } + + private async _appendBatchSkipDuplicates( + ops: Operation[], + source: 'local' | 'remote', + options: { pendingApply?: boolean } | undefined, + advanceSnapshotFrontier: boolean, ): Promise<{ seqs: number[]; writtenOps: Operation[]; skippedCount: number }> { if (ops.length === 0) { return { seqs: [], writtenOps: [], skippedCount: 0 }; @@ -795,11 +940,46 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort isFullStateOpType(op.opType))) { storeNames.push(STORE_NAMES.META); } await this._adapter.transaction(storeNames, 'readwrite', async (tx) => { + let stateCache: StateCacheEntry | undefined; + if (advanceSnapshotFrontier) { + stateCache = await tx.get( + STORE_NAMES.STATE_CACHE, + SINGLETON_KEY, + ); + if (!stateCache) { + throw new Error( + 'Cannot append snapshot-included operations without an existing state cache', + ); + } + + let preAppendLastSeq = 0; + await tx.iterate( + STORE_NAMES.OPS, + { direction: 'prev' }, + (_value, key) => { + if (typeof key !== 'number') { + throw new Error('Operation sequence key is not numeric'); + } + preAppendLastSeq = key; + return 'stop'; + }, + ); + if (stateCache.lastAppliedOpSeq !== preAppendLastSeq) { + throw new Error( + 'Cannot append snapshot-included operations when the state-cache frontier does not match the operation-log tail', + ); + } + } + + let lastIncludedSeq = 0; for (const op of ops) { // Check if op already exists in the same transaction (atomic) const existingKey = await tx.getKeyFromIndex( @@ -808,6 +988,10 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort 0) { diff --git a/src/app/op-log/persistence/sync-hydration.service.spec.ts b/src/app/op-log/persistence/sync-hydration.service.spec.ts index c2fe0c2577..ca6989398d 100644 --- a/src/app/op-log/persistence/sync-hydration.service.spec.ts +++ b/src/app/op-log/persistence/sync-hydration.service.spec.ts @@ -3,16 +3,19 @@ import { Store } from '@ngrx/store'; import { of } from 'rxjs'; import { SyncHydrationService } from './sync-hydration.service'; import { OperationLogStoreService } from './operation-log-store.service'; -import { StateSnapshotService } from '../backup/state-snapshot.service'; +import { AppStateSnapshot, StateSnapshotService } from '../backup/state-snapshot.service'; import { ClientIdService } from '../../core/util/client-id.service'; import { VectorClockService } from '../sync/vector-clock.service'; import { ValidateStateService } from '../validation/validate-state.service'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; -import { ActionType, OpType } from '../core/operation.types'; +import { ActionType, OperationLogEntry, OpType } from '../core/operation.types'; import { SyncProviderId } from '../sync-providers/provider.const'; import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const'; import { LOCAL_ONLY_SYNC_KEYS } from '../../features/config/local-only-sync-settings.util'; import { SnackService } from '../../core/snack/snack.service'; +import { ArchiveDbAdapter } from '../../core/persistence/archive-db-adapter.service'; +import { LockService } from '../sync/lock.service'; +import { LOCK_NAMES } from '../core/operation-log.const'; describe('SyncHydrationService', () => { let service: SyncHydrationService; @@ -23,6 +26,8 @@ describe('SyncHydrationService', () => { let mockVectorClockService: jasmine.SpyObj; let mockValidateStateService: jasmine.SpyObj; let mockSnackService: jasmine.SpyObj; + let mockArchiveDbAdapter: jasmine.SpyObj; + let mockLockService: jasmine.SpyObj; // Default local sync config for tests const defaultLocalSyncConfig = { @@ -42,6 +47,7 @@ describe('SyncHydrationService', () => { 'getLastSeq', 'saveStateCache', 'setVectorClock', + 'commitFileSnapshotBaseline', 'loadStateCache', 'getUnsynced', 'markRejected', @@ -66,6 +72,14 @@ describe('SyncHydrationService', () => { 'validateAndRepair', ]); mockSnackService = jasmine.createSpyObj('SnackService', ['open']); + mockArchiveDbAdapter = jasmine.createSpyObj('ArchiveDbAdapter', [ + 'saveArchiveYoung', + 'saveArchiveOld', + ]); + mockArchiveDbAdapter.saveArchiveYoung.and.resolveTo(); + mockArchiveDbAdapter.saveArchiveOld.and.resolveTo(); + mockLockService = jasmine.createSpyObj('LockService', ['request']); + mockLockService.request.and.callFake(async (_lockName, callback) => callback()); TestBed.configureTestingModule({ providers: [ @@ -77,6 +91,8 @@ describe('SyncHydrationService', () => { { provide: VectorClockService, useValue: mockVectorClockService }, { provide: ValidateStateService, useValue: mockValidateStateService }, { provide: SnackService, useValue: mockSnackService }, + { provide: ArchiveDbAdapter, useValue: mockArchiveDbAdapter }, + { provide: LockService, useValue: mockLockService }, ], }); service = TestBed.inject(SyncHydrationService); @@ -90,6 +106,11 @@ describe('SyncHydrationService', () => { mockOpLogStore.getLastSeq.and.resolveTo(10); mockOpLogStore.saveStateCache.and.resolveTo(undefined); mockOpLogStore.setVectorClock.and.resolveTo(undefined); + mockOpLogStore.commitFileSnapshotBaseline.and.resolveTo({ + seqs: [], + writtenOps: [], + skippedCount: 0, + }); mockValidateStateService.validateAndRepair.and.resolveTo({ isValid: true, wasRepaired: false, @@ -99,6 +120,38 @@ describe('SyncHydrationService', () => { describe('hydrateFromRemoteSync', () => { beforeEach(setupDefaultMocks); + it('serializes archive replacement and its snapshot read', async () => { + let isArchiveLockHeld = false; + mockLockService.request.and.callFake(async (lockName, callback) => { + expect(lockName).toBe(LOCK_NAMES.TASK_ARCHIVE); + isArchiveLockHeld = true; + try { + return await callback(); + } finally { + isArchiveLockHeld = false; + } + }); + mockArchiveDbAdapter.saveArchiveYoung.and.callFake(async () => { + expect(isArchiveLockHeld).toBeTrue(); + }); + mockArchiveDbAdapter.saveArchiveOld.and.callFake(async () => { + expect(isArchiveLockHeld).toBeTrue(); + }); + mockStateSnapshotService.getAllSyncModelDataFromStoreAsync.and.callFake( + async () => { + expect(isArchiveLockHeld).toBeTrue(); + return {} as AppStateSnapshot; + }, + ); + + await service.hydrateFromRemoteSync({ + archiveYoung: { task: { ids: [], entities: {} } }, + archiveOld: { task: { ids: [], entities: {} } }, + }); + + expect(mockLockService.request).toHaveBeenCalledTimes(1); + }); + it('should merge downloaded data with archive data from DB', async () => { const downloadedData = { task: { ids: ['t1'] }, project: { ids: ['p1'] } }; const archiveData = { @@ -456,24 +509,70 @@ describe('SyncHydrationService', () => { expect(mockOpLogStore.append).not.toHaveBeenCalled(); }); - it('should still save state cache when createSyncImportOp is false', async () => { + it('should reject only the pending ops captured before snapshot hydration starts', async () => { + const makeEntry = (id: string, seq: number): OperationLogEntry => ({ + seq, + op: { + id, + clientId: 'localClient', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: `task-${seq}`, + payload: { title: id }, + vectorClock: { localClient: seq }, + timestamp: seq, + schemaVersion: 1, + }, + appliedAt: seq, + source: 'local', + }); + const pendingBeforeHydration = makeEntry('pending-before-hydration', 1); + const pendingDuringHydration = makeEntry('pending-during-hydration', 2); + let snapshotReadStarted = false; + mockOpLogStore.getUnsynced.and.callFake(async () => + snapshotReadStarted + ? [pendingBeforeHydration, pendingDuringHydration] + : [pendingBeforeHydration], + ); + mockStateSnapshotService.getAllSyncModelDataFromStoreAsync.and.callFake( + async () => { + snapshotReadStarted = true; + return {} as never; + }, + ); + + await service.hydrateFromRemoteSync({ task: {} }, undefined, false); + + expect(mockOpLogStore.markRejected).toHaveBeenCalledOnceWith([ + pendingBeforeHydration.op.id, + ]); + }); + + it('should commit the file snapshot baseline when createSyncImportOp is false', async () => { mockOpLogStore.getLastSeq.and.resolveTo(42); await service.hydrateFromRemoteSync({ task: {} }, undefined, false); - expect(mockOpLogStore.saveStateCache).toHaveBeenCalled(); + expect(mockOpLogStore.commitFileSnapshotBaseline).toHaveBeenCalledWith( + jasmine.objectContaining({ lastAppliedOpSeq: 42 }), + ); + expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); }); - it('should still update vector clock when createSyncImportOp is false', async () => { + it('should include the vector clock in the atomic file baseline', async () => { mockVectorClockService.getCurrentVectorClock.and.resolveTo({ localClient: 5 }); await service.hydrateFromRemoteSync({ task: {} }, undefined, false); - expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith( + expect(mockOpLogStore.commitFileSnapshotBaseline).toHaveBeenCalledWith( jasmine.objectContaining({ - localClient: 6, // Should still increment + vectorClock: jasmine.objectContaining({ + localClient: 6, + }), }), ); + expect(mockOpLogStore.setVectorClock).not.toHaveBeenCalled(); }); it('should still dispatch loadAllData when createSyncImportOp is false', async () => { @@ -490,16 +589,178 @@ describe('SyncHydrationService', () => { ); }); + it('should invoke beforeStateLoad immediately before replacing NgRx state', async () => { + let dispatchCountAtHook = -1; + + await service.hydrateFromRemoteSync({}, undefined, false, undefined, { + beforeStateLoad: () => { + dispatchCountAtHook = mockStore.dispatch.calls.count(); + }, + }); + + expect(dispatchCountAtHook).toBe(0); + expect(mockStore.dispatch).toHaveBeenCalledTimes(1); + }); + + it('should invoke afterStateLoad only after loadAllData dispatch commits', async () => { + let dispatchCountBeforeStateLoad = -1; + let dispatchCountAfterStateLoad = -1; + + await service.hydrateFromRemoteSync({}, undefined, false, undefined, { + beforeStateLoad: () => { + dispatchCountBeforeStateLoad = mockStore.dispatch.calls.count(); + }, + afterStateLoad: () => { + dispatchCountAfterStateLoad = mockStore.dispatch.calls.count(); + }, + }); + + expect(dispatchCountBeforeStateLoad).toBe(0); + expect(dispatchCountAfterStateLoad).toBe(1); + }); + + it('should not invoke afterStateLoad when loadAllData dispatch throws', async () => { + let didRunBeforeStateLoad = false; + let didRunAfterStateLoad = false; + mockStore.dispatch.and.throwError('state dispatch failed'); + + await expectAsync( + service.hydrateFromRemoteSync({}, undefined, false, undefined, { + beforeStateLoad: () => { + didRunBeforeStateLoad = true; + }, + afterStateLoad: () => { + didRunAfterStateLoad = true; + }, + }), + ).toBeRejectedWithError('state dispatch failed'); + + expect(didRunBeforeStateLoad).toBeTrue(); + expect(didRunAfterStateLoad).toBeFalse(); + }); + + it('should signal snapshot persistence only after cache and vector clock commit', async () => { + const callOrder: string[] = []; + mockOpLogStore.commitFileSnapshotBaseline.and.callFake(async () => { + callOrder.push('commit-snapshot-baseline'); + return { seqs: [], writtenOps: [], skippedCount: 0 }; + }); + + await service.hydrateFromRemoteSync({}, undefined, false, undefined, { + afterSnapshotCachePersisted: () => { + callOrder.push('after-snapshot-cache-persisted'); + }, + afterSnapshotPersisted: () => { + callOrder.push('after-snapshot-persisted'); + }, + beforeStateLoad: () => { + callOrder.push('before-state-load'); + }, + }); + + expect(callOrder).toEqual([ + 'commit-snapshot-baseline', + 'after-snapshot-cache-persisted', + 'after-snapshot-persisted', + 'before-state-load', + ]); + }); + + it('should not signal or dispatch when the atomic snapshot baseline fails', async () => { + let didPersistSnapshotCache = false; + let didPersistSnapshot = false; + let didRunBeforeStateLoad = false; + let didRunAfterStateLoad = false; + mockOpLogStore.commitFileSnapshotBaseline.and.rejectWith( + new Error('snapshot baseline write failed'), + ); + + await expectAsync( + service.hydrateFromRemoteSync({}, undefined, false, undefined, { + afterSnapshotCachePersisted: () => { + didPersistSnapshotCache = true; + }, + afterSnapshotPersisted: () => { + didPersistSnapshot = true; + }, + beforeStateLoad: () => { + didRunBeforeStateLoad = true; + }, + afterStateLoad: () => { + didRunAfterStateLoad = true; + }, + }), + ).toBeRejectedWithError('snapshot baseline write failed'); + + expect(didPersistSnapshotCache).toBeFalse(); + expect(didPersistSnapshot).toBeFalse(); + expect(didRunBeforeStateLoad).toBeFalse(); + expect(didRunAfterStateLoad).toBeFalse(); + expect(mockStore.dispatch).not.toHaveBeenCalled(); + }); + + it('should signal archive replacement after the atomic baseline commits', async () => { + const callOrder: string[] = []; + mockOpLogStore.commitFileSnapshotBaseline.and.callFake(async () => { + callOrder.push('commit-snapshot-baseline'); + return { seqs: [], writtenOps: [], skippedCount: 0 }; + }); + + await service.hydrateFromRemoteSync( + { + task: {}, + archiveYoung: { task: { ids: [], entities: {} } }, + }, + undefined, + false, + undefined, + { + afterArchiveReplacement: () => callOrder.push('after-archive-replace'), + }, + ); + + expect(callOrder).toEqual(['commit-snapshot-baseline', 'after-archive-replace']); + expect(mockArchiveDbAdapter.saveArchiveYoung).not.toHaveBeenCalled(); + }); + + it('should not signal archive replacement when the atomic baseline fails', async () => { + let didReplaceArchive = false; + mockOpLogStore.commitFileSnapshotBaseline.and.rejectWith( + new Error('snapshot baseline write failed'), + ); + + await expectAsync( + service.hydrateFromRemoteSync( + { + task: {}, + archiveYoung: { task: { ids: [], entities: {} } }, + }, + undefined, + false, + undefined, + { + afterArchiveReplacement: () => { + didReplaceArchive = true; + }, + }, + ), + ).toBeRejectedWithError('snapshot baseline write failed'); + + expect(didReplaceArchive).toBeFalse(); + }); + it('should still merge remote vector clock when createSyncImportOp is false', async () => { const remoteVectorClock = { remoteClient: 100 }; mockVectorClockService.getCurrentVectorClock.and.resolveTo({ localClient: 5 }); await service.hydrateFromRemoteSync({ task: {} }, remoteVectorClock, false); - expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith( + expect(mockOpLogStore.commitFileSnapshotBaseline).toHaveBeenCalledWith( jasmine.objectContaining({ - localClient: 6, - remoteClient: 100, + vectorClock: jasmine.objectContaining({ + localClient: 6, + remoteClient: 100, + }), }), ); }); diff --git a/src/app/op-log/persistence/sync-hydration.service.ts b/src/app/op-log/persistence/sync-hydration.service.ts index b377b84806..752f5f0f61 100644 --- a/src/app/op-log/persistence/sync-hydration.service.ts +++ b/src/app/op-log/persistence/sync-hydration.service.ts @@ -28,6 +28,23 @@ import { applyLocalOnlySyncSettingsToAppData, stripLocalOnlySyncSettingsFromAppData, } from '../../features/config/local-only-sync-settings.util'; +import { LockService } from '../sync/lock.service'; +import { LOCK_NAMES } from '../core/operation-log.const'; + +interface SnapshotHydrationHooks { + /** Remote operations already represented by a file-based snapshot. */ + snapshotIncludedOps?: readonly Operation[]; + /** Runs synchronously after downloaded archive replacement commits. */ + afterArchiveReplacement?: () => void; + /** Runs synchronously after the snapshot baseline transaction commits. */ + afterSnapshotCachePersisted?: () => void; + /** Runs synchronously after the complete snapshot baseline commits. */ + afterSnapshotPersisted?: () => void; + /** Runs synchronously immediately before loadAllData replaces live NgRx state. */ + beforeStateLoad?: () => void; + /** Runs synchronously after loadAllData has replaced live NgRx state. */ + afterStateLoad?: () => void; +} /** * Handles hydration after remote sync downloads. @@ -52,6 +69,7 @@ export class SyncHydrationService { private sessionValidation = inject(SyncSessionValidationService); private snackService = inject(SnackService); private archiveDbAdapter = inject(ArchiveDbAdapter); + private lockService = inject(LockService); /** * Handles hydration after a remote sync download. @@ -71,16 +89,27 @@ export class SyncHydrationService { * for file-based sync bootstrap to avoid "clean slate" semantics that would filter * concurrent ops from other clients. Default is true for backwards compatibility * and for explicit "use local/remote" conflict resolution flows. + * @param hooks - Internal orchestration hooks around archive and state replacement. */ async hydrateFromRemoteSync( downloadedMainModelData?: Record, remoteVectorClock?: Record, createSyncImportOp: boolean = true, syncImportReason?: SyncImportReason, + hooks?: SnapshotHydrationHooks, ): Promise { OpLog.normal('SyncHydrationService: Hydrating from remote sync...'); try { + // Capture the exact pending set before any snapshot work can yield. The + // normal file-snapshot caller opens a remote-apply window around this + // method, so user actions arriving after this read are deferred and must + // be preserved on top of the downloaded snapshot rather than rejected as + // if they belonged to the superseded local baseline. + const unsyncedOpsToReject = createSyncImportOp + ? [] + : await this.opLogStore.getUnsynced(); + // 0. Capture current local-only sync settings BEFORE overwriting // These settings should remain local to each client and not be overwritten by remote data. // FIX: isEnabled was not being preserved, causing sync to appear disabled after reload @@ -94,31 +123,45 @@ export class SyncHydrationService { isManualSyncOnly: currentSyncConfig.isManualSyncOnly, }; - // 1. Write archives to IndexedDB if they were included in the downloaded data. - // This is critical for file-based sync where archives are bundled with the snapshot. - // Archives must be written BEFORE we read them back via getAllSyncModelDataFromStoreAsync(). - // NOTE: This only happens when actually applying remote state (not during conflict - // detection - the downloadOps method no longer writes archives prematurely). - if (downloadedMainModelData) { - const typedData = downloadedMainModelData as Record; - if (typedData['archiveYoung']) { - await this.archiveDbAdapter.saveArchiveYoung( - typedData['archiveYoung'] as ArchiveModel, - ); - OpLog.normal('SyncHydrationService: Wrote archiveYoung to IndexedDB from sync'); - } - if (typedData['archiveOld']) { - await this.archiveDbAdapter.saveArchiveOld( - typedData['archiveOld'] as ArchiveModel, - ); - OpLog.normal('SyncHydrationService: Wrote archiveOld to IndexedDB from sync'); - } - } + const typedDownloadedData = downloadedMainModelData as + | Record + | undefined; + const downloadedArchiveYoung = typedDownloadedData?.['archiveYoung'] as + | ArchiveModel + | undefined; + const downloadedArchiveOld = typedDownloadedData?.['archiveOld'] as + | ArchiveModel + | undefined; - // 2. Read archive data from IndexedDB and merge with passed entity data - // Entity models (task, tag, project, etc.) come from downloadedMainModelData - // Archive models (archiveYoung, archiveOld) come from IndexedDB (now with synced data) - const dbData = await this.stateSnapshotService.getAllSyncModelDataFromStoreAsync(); + // 1. Replace downloaded archives and read the resulting snapshot under one + // archive lock. Otherwise a local archive read-modify-write can start from + // the old archive, then save it after this replacement and silently erase + // downloaded entries. TASK_ARCHIVE is independent from OPERATION_LOG. + const dbData = await this.lockService.request(LOCK_NAMES.TASK_ARCHIVE, async () => { + // Full-state imports retain their existing archive write order. File + // snapshots defer these writes to commitFileSnapshotBaseline(), where + // archives, state, clock, and included operations commit atomically. + if (createSyncImportOp) { + if (downloadedArchiveYoung) { + await this.archiveDbAdapter.saveArchiveYoung(downloadedArchiveYoung); + hooks?.afterArchiveReplacement?.(); + OpLog.normal( + 'SyncHydrationService: Wrote archiveYoung to IndexedDB from sync', + ); + } + if (downloadedArchiveOld) { + await this.archiveDbAdapter.saveArchiveOld(downloadedArchiveOld); + hooks?.afterArchiveReplacement?.(); + OpLog.normal('SyncHydrationService: Wrote archiveOld to IndexedDB from sync'); + } + } + + // Archives must be read after the optional replacement while the same + // lock is still held so the state cache and loadAllData use one view. + return this.stateSnapshotService.getAllSyncModelDataFromStoreAsync(); + }); + + // 2. Merge the serialized archive data with passed entity data. const mergedData = downloadedMainModelData ? { ...dbData, ...downloadedMainModelData } @@ -207,12 +250,11 @@ export class SyncHydrationService { // CRITICAL: Reject any local pending ops since they're now based on superseded state. // Without SYNC_IMPORT, SyncImportFilterService won't automatically filter them. // These ops have superseded clocks and payloads that don't match the new snapshot. - const unsyncedOps = await this.opLogStore.getUnsynced(); - if (unsyncedOps.length > 0) { - const opIds = unsyncedOps.map((entry) => entry.op.id); + if (unsyncedOpsToReject.length > 0) { + const opIds = unsyncedOpsToReject.map((entry) => entry.op.id); await this.opLogStore.markRejected(opIds); OpLog.normal( - `SyncHydrationService: Rejected ${unsyncedOps.length} local pending op(s) ` + + `SyncHydrationService: Rejected ${unsyncedOpsToReject.length} local pending op(s) ` + `(superseded after file-based sync snapshot)`, ); @@ -220,7 +262,7 @@ export class SyncHydrationService { this.snackService.open({ msg: T.F.SYNC.S.LOCAL_CHANGES_DISCARDED_SNAPSHOT, translateParams: { - count: unsyncedOps.length, + count: unsyncedOpsToReject.length, }, }); } @@ -280,25 +322,52 @@ export class SyncHydrationService { clockForStorage = newClock; } - // 9. Save new state cache (snapshot) for crash safety - await this.opLogStore.saveStateCache({ - state: dataToLoad, - lastAppliedOpSeq: lastSeq, - vectorClock: clockForStorage, - compactedAt: Date.now(), - }); - OpLog.normal('SyncHydrationService: Saved state cache after sync'); + // 9. Commit the durable snapshot baseline before replacing live state. + // File snapshots include the downloaded archives and represented remote + // operations in this same transaction. If any write fails, the old + // baseline remains intact and mid-hydration local actions can safely drain + // against it; there is no cache-only or ops-only restart state. + if (createSyncImportOp) { + await this.opLogStore.saveStateCache({ + state: dataToLoad, + lastAppliedOpSeq: lastSeq, + vectorClock: clockForStorage, + compactedAt: Date.now(), + }); + hooks?.afterSnapshotCachePersisted?.(); - // 10. Update vector clock store - // This is critical because: - // - The SYNC_IMPORT was appended with source='remote', so store wasn't updated - // - If user creates new ops in this session, incrementAndStoreVectorClock reads from store - // - Without this, new ops would have clocks missing entries from the SYNC_IMPORT - await this.opLogStore.setVectorClock(clockForStorage); - OpLog.normal('SyncHydrationService: Updated vector clock store after sync'); + // The SYNC_IMPORT was appended with source='remote', so update the + // working clock separately on this legacy full-state-import path. + await this.opLogStore.setVectorClock(clockForStorage); + } else { + const appendResult = await this.opLogStore.commitFileSnapshotBaseline({ + state: dataToLoad, + lastAppliedOpSeq: lastSeq, + vectorClock: clockForStorage, + compactedAt: Date.now(), + snapshotIncludedOps: hooks?.snapshotIncludedOps ?? [], + ...(downloadedArchiveYoung ? { archiveYoung: downloadedArchiveYoung } : {}), + ...(downloadedArchiveOld ? { archiveOld: downloadedArchiveOld } : {}), + }); + if (downloadedArchiveYoung || downloadedArchiveOld) { + hooks?.afterArchiveReplacement?.(); + } + hooks?.afterSnapshotCachePersisted?.(); + OpLog.normal( + `SyncHydrationService: Atomically committed file snapshot and ` + + `${appendResult.writtenOps.length} included operation(s)` + + (appendResult.skippedCount > 0 + ? `; skipped ${appendResult.skippedCount} duplicate(s)` + : ''), + ); + } + hooks?.afterSnapshotPersisted?.(); + OpLog.normal('SyncHydrationService: Committed snapshot persistence baseline'); - // 11. Dispatch loadAllData to update NgRx + // 10. Dispatch loadAllData to update NgRx + hooks?.beforeStateLoad?.(); this.store.dispatch(loadAllData({ appDataComplete: dataToLoad })); + hooks?.afterStateLoad?.(); OpLog.normal('SyncHydrationService: Dispatched loadAllData with synced data'); } catch (e) { OpLog.err('SyncHydrationService: Error during hydrateFromRemoteSync', e); 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 3a737277c1..d61edf097f 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 @@ -7,7 +7,15 @@ import { SnackService } from '../../core/snack/snack.service'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { VectorClockService } from './vector-clock.service'; import { OperationApplierService } from '../apply/operation-applier.service'; +import { HydrationStateService } from '../apply/hydration-state.service'; import { OperationLogEffects } from '../capture/operation-log.effects'; +import { + acknowledgeDeferredAction, + bufferDeferredAction, + clearDeferredActions, + getDeferredActions, +} from '../capture/operation-capture.meta-reducer'; +import { PersistentAction } from '../core/persistent-action.interface'; import { ConflictResolutionService } from './conflict-resolution.service'; import { ValidateStateService } from '../validation/validate-state.service'; import { SyncSessionValidationService } from './sync-session-validation.service'; @@ -73,6 +81,7 @@ describe('OperationLogSyncService', () => { let validateStateServiceSpy: jasmine.SpyObj; let lockServiceSpy: jasmine.SpyObj; let operationApplierSpy: jasmine.SpyObj; + let hydrationStateServiceSpy: jasmine.SpyObj; let operationLogEffectsSpy: jasmine.SpyObj; let hydratorServiceSpy: jasmine.SpyObj; const defaultBackupRef = { backupId: 'backup-1', savedAt: 1 }; @@ -117,6 +126,7 @@ describe('OperationLogSyncService', () => { 'clearFullStateOps', 'getVectorClock', 'appendBatchSkipDuplicates', + 'appendSnapshotIncludedOps', 'hasSyncedOps', 'runRemoteStateReplacement', 'isRawRebuildIncomplete', @@ -140,6 +150,11 @@ describe('OperationLogSyncService', () => { writtenOps: [], skippedCount: 0, }); + opLogStoreSpy.appendSnapshotIncludedOps.and.resolveTo({ + seqs: [], + writtenOps: [], + skippedCount: 0, + }); opLogStoreSpy.runRemoteStateReplacement.and.resolveTo(); opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(false); opLogStoreSpy.loadRawRebuildIncomplete.and.resolveTo(null); @@ -244,10 +259,18 @@ describe('OperationLogSyncService', () => { 'applyOperations', ]); operationApplierSpy.applyOperations.and.resolveTo({ appliedOps: [] }); + hydrationStateServiceSpy = jasmine.createSpyObj('HydrationStateService', [ + 'startApplyingRemoteOps', + 'endApplyingRemoteOps', + ]); operationLogEffectsSpy = jasmine.createSpyObj('OperationLogEffects', [ 'processDeferredActions', ]); - operationLogEffectsSpy.processDeferredActions.and.resolveTo(); + operationLogEffectsSpy.processDeferredActions.and.callFake(async () => { + for (const action of getDeferredActions()) { + acknowledgeDeferredAction(action); + } + }); hydratorServiceSpy = jasmine.createSpyObj('OperationLogHydratorService', [ 'retryFailedRemoteOps', ]); @@ -273,6 +296,7 @@ describe('OperationLogSyncService', () => { provide: OperationApplierService, useValue: operationApplierSpy, }, + { provide: HydrationStateService, useValue: hydrationStateServiceSpy }, { provide: OperationLogEffects, useValue: operationLogEffectsSpy }, { provide: OperationLogHydratorService, useValue: hydratorServiceSpy }, { @@ -1754,6 +1778,1005 @@ describe('OperationLogSyncService', () => { }); describe('LocalDataConflictError for file-based sync', () => { + it('should abort snapshot hydration when a local op becomes durable after conflict detection', async () => { + const lateLocalEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'late-local-op', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-late', + payload: { task: { id: 'task-late', changes: { title: 'Keep me' } } }, + vectorClock: { clientA: 2 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([lateLocalEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError(LocalDataConflictError); + + expect(syncHydrationServiceSpy.hydrateFromRemoteSync).not.toHaveBeenCalled(); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + }); + + it('should restore and persist a local action buffered during snapshot hydration', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const actionAfterStateLoad: PersistentAction = { + ...localAction, + task: { id: 'task-after-load', changes: { title: 'Already restored' } }, + meta: { ...localAction.meta, entityId: 'task-after-load' }, + }; + const persistedEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'buffered-local-op', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-local', + payload: { task: localAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + const persistedAfterLoadEntry: OperationLogEntry = { + ...persistedEntry, + seq: 3, + op: { + ...persistedEntry.op, + id: 'after-load-local-op', + entityId: 'task-after-load', + payload: { task: actionAfterStateLoad.task }, + vectorClock: { clientA: 3, clientB: 5 }, + timestamp: 3, + }, + appliedAt: 3, + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([persistedEntry, persistedAfterLoadEntry]), + Promise.resolve([persistedEntry, persistedAfterLoadEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + bufferDeferredAction(localAction); + hooks?.afterArchiveReplacement?.(); + hooks?.beforeStateLoad?.(); + bufferDeferredAction(actionAfterStateLoad); + hooks?.afterStateLoad?.(); + }, + ); + operationApplierSpy.applyOperations.and.resolveTo({ + appliedOps: [persistedEntry.op, persistedAfterLoadEntry.op], + }); + const mockStore = TestBed.inject(MockStore); + const dispatchSpy = spyOn(mockStore, 'dispatch').and.callThrough(); + const durabilityOrder: string[] = []; + operationLogEffectsSpy.processDeferredActions.and.callFake(async () => { + if ( + getDeferredActions().includes(localAction) && + !durabilityOrder.includes('persist') + ) { + const wasAlreadyReplayed = dispatchSpy.calls + .allArgs() + .some(([dispatched]) => { + const action = dispatched as unknown; + return ( + typeof action === 'object' && + action !== null && + 'type' in action && + action.type === localAction.type && + 'meta' in action && + typeof action.meta === 'object' && + action.meta !== null && + 'isRemote' in action.meta && + action.meta.isRemote === true + ); + }); + durabilityOrder.push( + wasAlreadyReplayed ? 'replay-before-persist' : 'persist', + ); + } + for (const action of getDeferredActions()) { + acknowledgeDeferredAction(action); + } + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await service.downloadRemoteOps(mockProvider); + + expect(hydrationStateServiceSpy.startApplyingRemoteOps).toHaveBeenCalledTimes( + 1, + ); + expect(hydrationStateServiceSpy.endApplyingRemoteOps).toHaveBeenCalledTimes( + 1, + ); + expect(dispatchSpy).toHaveBeenCalledWith( + jasmine.objectContaining({ + type: localAction.type, + meta: jasmine.objectContaining({ isRemote: true }), + }), + ); + const remotelyReplayedTaskIds = dispatchSpy.calls + .allArgs() + .map(([action]) => action as unknown) + .filter( + ( + action, + ): action is { + task?: { id?: string }; + meta?: { isRemote?: boolean }; + } => typeof action === 'object' && action !== null, + ) + .filter((action) => action.meta?.isRemote) + .map((action) => action.task?.id); + expect(remotelyReplayedTaskIds).toContain('task-local'); + expect(remotelyReplayedTaskIds).not.toContain('task-after-load'); + expect(durabilityOrder).toEqual(['persist']); + expect(operationLogEffectsSpy.processDeferredActions.calls.allArgs()).toEqual( + [ + [{ callerHoldsOperationLogLock: false }], + [{ callerHoldsOperationLogLock: true }], + ], + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledOnceWith( + [persistedEntry.op, persistedAfterLoadEntry.op], + { + isLocalHydration: false, + skipDeferredLocalActions: true, + skipReducerDispatch: true, + remoteApplyWindowAlreadyOpen: true, + }, + ); + expect(mockProvider.setLastServerSeq).toHaveBeenCalledOnceWith(1); + } finally { + clearDeferredActions(); + } + }); + + it('should persist and restore an action that arrives during the final deferred drain', async () => { + clearDeferredActions(); + const lateAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-late', changes: { title: 'Keep me too' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-late', + opType: OpType.Update, + }, + }; + const lateEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'late-during-final-drain', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-late', + payload: { task: lateAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + let hydrationFinished = false; + let lateActionPersisted = false; + opLogStoreSpy.getUnsynced.and.callFake(async () => + hydrationFinished && lateActionPersisted ? [lateEntry] : [], + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + hooks?.afterArchiveReplacement?.(); + hooks?.beforeStateLoad?.(); + hooks?.afterStateLoad?.(); + hydrationFinished = true; + }, + ); + const callOrder: string[] = []; + let heldLockDrainCount = 0; + operationLogEffectsSpy.processDeferredActions.and.callFake(async (options) => { + if (!options?.callerHoldsOperationLogLock) return; + + heldLockDrainCount++; + if (heldLockDrainCount === 1) { + // processDeferredActions snapshots the queue before awaiting its + // writes. This action therefore belongs to the next drain. + bufferDeferredAction(lateAction); + callOrder.push('late-action-buffered'); + return; + } + + lateActionPersisted = true; + acknowledgeDeferredAction(lateAction); + callOrder.push('late-action-persisted'); + }); + operationApplierSpy.applyOperations.and.callFake(async (ops) => { + callOrder.push('late-archive-restored'); + return { appliedOps: ops }; + }); + hydrationStateServiceSpy.endApplyingRemoteOps.and.callFake(() => { + callOrder.push('remote-window-closed'); + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await service.downloadRemoteOps(mockProvider); + + expect(heldLockDrainCount).toBe(2); + expect(getDeferredActions()).not.toContain(lateAction); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledOnceWith( + [lateEntry.op], + { + isLocalHydration: false, + skipDeferredLocalActions: true, + skipReducerDispatch: true, + remoteApplyWindowAlreadyOpen: true, + }, + ); + expect(callOrder).toEqual([ + 'late-action-buffered', + 'late-action-persisted', + 'late-archive-restored', + 'remote-window-closed', + ]); + } finally { + clearDeferredActions(); + } + }); + + it('should commit snapshot-included remote ops BEFORE persisting deferred local intents', async () => { + clearDeferredActions(); + const snapshotIncludedOp: Operation = { + id: 'snap-op-1', + clientId: 'client-B', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-x', + payload: { task: { id: 'task-x', changes: {} } }, + vectorClock: { clientB: 4 }, + timestamp: 1, + schemaVersion: 1, + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [snapshotIncludedOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + snapshotAppliedOpIds: ['snap-op-1'], + latestServerSeq: 1, + }); + const callOrder: string[] = []; + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + expect(hooks?.snapshotIncludedOps).toEqual([snapshotIncludedOp]); + callOrder.push('commit-snapshot-baseline'); + }, + ); + operationLogEffectsSpy.processDeferredActions.and.callFake( + async (options?: { callerHoldsOperationLogLock?: boolean }) => { + if (options?.callerHoldsOperationLogLock) { + callOrder.push('persist-deferred'); + } + }, + ); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await service.downloadRemoteOps(mockProvider); + + // Frontier ordering: getEntityFrontier() takes the LAST op per entity + // in seq order, so the snapshot's (older) remote ops must land at + // lower seqs than any local intents persisted during hydration — + // otherwise the frontier regresses and a later concurrent remote op + // is misclassified as non-conflicting. + expect(callOrder[0]).toBe('commit-snapshot-baseline'); + expect(callOrder).toContain('persist-deferred'); + expect(callOrder.indexOf('commit-snapshot-baseline')).toBeLessThan( + callOrder.indexOf('persist-deferred'), + ); + }); + + it('should persist deferred local intents against the old baseline when the atomic snapshot commit fails', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const snapshotIncludedOp: Operation = { + id: 'snap-op-append-failure', + clientId: 'client-B', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-x', + payload: { task: { id: 'task-x', changes: {} } }, + vectorClock: { clientB: 4 }, + timestamp: 1, + schemaVersion: 1, + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [snapshotIncludedOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + snapshotAppliedOpIds: [snapshotIncludedOp.id], + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + expect(hooks?.snapshotIncludedOps).toEqual([snapshotIncludedOp]); + bufferDeferredAction(localAction); + throw new Error('snapshot baseline write failed'); + }, + ); + const callOrder: string[] = []; + hydrationStateServiceSpy.endApplyingRemoteOps.and.callFake(() => { + callOrder.push('end-remote-window'); + }); + operationLogEffectsSpy.processDeferredActions.and.callFake(async (options) => { + if (options?.callerHoldsOperationLogLock) { + callOrder.push('persist-deferred'); + acknowledgeDeferredAction(localAction); + } + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('snapshot baseline write failed'); + + expect(operationLogEffectsSpy.processDeferredActions.calls.allArgs()).toEqual( + [ + [{ callerHoldsOperationLogLock: false }], + [{ callerHoldsOperationLogLock: true }], + ], + ); + expect(callOrder).toEqual(['persist-deferred', 'end-remote-window']); + expect(getDeferredActions()).not.toContain(localAction); + expect(operationApplierSpy.applyOperations).not.toHaveBeenCalled(); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + + it('should append snapshot ops before recovery when persistence completes before state dispatch', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const snapshotIncludedOp: Operation = { + id: 'snap-op-persisted-before-dispatch-failure', + clientId: 'client-B', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-x', + payload: { task: { id: 'task-x', changes: {} } }, + vectorClock: { clientB: 4 }, + timestamp: 1, + schemaVersion: 1, + }; + const persistedEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'local-op-persisted-before-dispatch-failure', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-local', + payload: { task: localAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([persistedEntry]), + Promise.resolve([persistedEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [snapshotIncludedOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + snapshotAppliedOpIds: [snapshotIncludedOp.id], + latestServerSeq: 1, + }); + const callOrder: string[] = []; + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + bufferDeferredAction(localAction); + callOrder.push('commit-snapshot-baseline'); + hooks?.afterArchiveReplacement?.(); + hooks?.afterSnapshotCachePersisted?.(); + hooks?.afterSnapshotPersisted?.(); + hooks?.beforeStateLoad?.(); + throw new Error('failed before state dispatch'); + }, + ); + operationLogEffectsSpy.processDeferredActions.and.callFake(async (options) => { + if (options?.callerHoldsOperationLogLock) { + callOrder.push('persist-deferred'); + acknowledgeDeferredAction(localAction); + } + }); + operationApplierSpy.applyOperations.and.resolveTo({ + appliedOps: [persistedEntry.op], + }); + const dispatchSpy = spyOn( + TestBed.inject(MockStore), + 'dispatch', + ).and.callThrough(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('failed before state dispatch'); + + expect(callOrder[0]).toBe('commit-snapshot-baseline'); + expect(callOrder).toContain('persist-deferred'); + expect(dispatchSpy).not.toHaveBeenCalledWith( + jasmine.objectContaining({ + type: localAction.type, + meta: jasmine.objectContaining({ isRemote: true }), + }), + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith( + [persistedEntry.op], + jasmine.objectContaining({ skipReducerDispatch: true }), + ); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + + it('should persist deferred intents when a vector write aborts the atomic snapshot transaction', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const snapshotIncludedOp: Operation = { + id: 'snap-op-partial-persistence', + clientId: 'client-B', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-x', + payload: { task: { id: 'task-x', changes: {} } }, + vectorClock: { clientB: 4 }, + timestamp: 1, + schemaVersion: 1, + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [snapshotIncludedOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + snapshotAppliedOpIds: [snapshotIncludedOp.id], + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + expect(hooks?.snapshotIncludedOps).toEqual([snapshotIncludedOp]); + bufferDeferredAction(localAction); + throw new Error('atomic vector clock write failed'); + }, + ); + operationLogEffectsSpy.processDeferredActions.and.callFake(async (options) => { + if (options?.callerHoldsOperationLogLock) { + acknowledgeDeferredAction(localAction); + } + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('atomic vector clock write failed'); + + expect(operationLogEffectsSpy.processDeferredActions.calls.allArgs()).toEqual( + [ + [{ callerHoldsOperationLogLock: false }], + [{ callerHoldsOperationLogLock: true }], + ], + ); + expect(operationApplierSpy.applyOperations).not.toHaveBeenCalled(); + expect(getDeferredActions()).not.toContain(localAction); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + + it('should persist buffered local actions when snapshot hydration fails', async () => { + clearDeferredActions(); + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.rejectWith( + new Error('hydration boom'), + ); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('hydration boom'); + + // The remote-apply window must be closed AND the deferred buffer + // persisted: edits made during the failed hydration are applied to + // live NgRx state but would otherwise be silently lost on app exit. + expect(hydrationStateServiceSpy.endApplyingRemoteOps).toHaveBeenCalled(); + expect(operationLogEffectsSpy.processDeferredActions).toHaveBeenCalledWith({ + callerHoldsOperationLogLock: true, + }); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + }); + + it('should restore overwritten reducers and archives when hydration fails after state load', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const persistedEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'buffered-local-op-after-failure', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-local', + payload: { task: localAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([persistedEntry]), + Promise.resolve([persistedEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + bufferDeferredAction(localAction); + hooks?.afterArchiveReplacement?.(); + hooks?.beforeStateLoad?.(); + hooks?.afterStateLoad?.(); + throw new Error('state load failed'); + }, + ); + operationApplierSpy.applyOperations.and.resolveTo({ + appliedOps: [persistedEntry.op], + }); + const dispatchSpy = spyOn( + TestBed.inject(MockStore), + 'dispatch', + ).and.callThrough(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('state load failed'); + + expect(dispatchSpy).toHaveBeenCalledWith( + jasmine.objectContaining({ + type: localAction.type, + meta: jasmine.objectContaining({ isRemote: true }), + }), + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith( + [persistedEntry.op], + jasmine.objectContaining({ + skipReducerDispatch: true, + skipDeferredLocalActions: true, + remoteApplyWindowAlreadyOpen: true, + }), + ); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + + it('should restore archives without replaying reducers when state load does not commit', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const persistedEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'buffered-local-op-after-archive-replacement', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-local', + payload: { task: localAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([persistedEntry]), + Promise.resolve([persistedEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + bufferDeferredAction(localAction); + hooks?.afterArchiveReplacement?.(); + hooks?.beforeStateLoad?.(); + throw new Error('state load failed before commit'); + }, + ); + operationApplierSpy.applyOperations.and.resolveTo({ + appliedOps: [persistedEntry.op], + }); + const dispatchSpy = spyOn( + TestBed.inject(MockStore), + 'dispatch', + ).and.callThrough(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await expectAsync( + service.downloadRemoteOps(mockProvider), + ).toBeRejectedWithError('state load failed before commit'); + + expect(dispatchSpy).not.toHaveBeenCalledWith( + jasmine.objectContaining({ + type: localAction.type, + meta: jasmine.objectContaining({ isRemote: true }), + }), + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith( + [persistedEntry.op], + jasmine.objectContaining({ + skipReducerDispatch: true, + skipDeferredLocalActions: true, + remoteApplyWindowAlreadyOpen: true, + }), + ); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + } finally { + clearDeferredActions(); + } + }); + + it('should continue reducer and archive restoration after a transient deferred drain failure', async () => { + clearDeferredActions(); + const localAction: PersistentAction = { + type: ActionType.TASK_SHARED_UPDATE, + task: { id: 'task-local', changes: { title: 'Keep me' } }, + meta: { + isPersistent: true, + entityType: 'TASK', + entityId: 'task-local', + opType: OpType.Update, + }, + }; + const persistedEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'buffered-local-op-after-drain-retry', + clientId: 'client-A', + actionType: ActionType.TASK_SHARED_UPDATE, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-local', + payload: { task: localAction.task }, + vectorClock: { clientA: 2, clientB: 5 }, + timestamp: 2, + schemaVersion: 1, + }, + appliedAt: 2, + source: 'local', + }; + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([]), + Promise.resolve([persistedEntry]), + Promise.resolve([persistedEntry]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['remote-task'] } }, + snapshotVectorClock: { clientB: 5 }, + latestServerSeq: 1, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + syncHydrationServiceSpy.hydrateFromRemoteSync.and.callFake( + async (_state, _clock, _createImport, _reason, hooks) => { + bufferDeferredAction(localAction); + hooks?.afterArchiveReplacement?.(); + hooks?.beforeStateLoad?.(); + hooks?.afterStateLoad?.(); + }, + ); + let heldLockDrainCount = 0; + operationLogEffectsSpy.processDeferredActions.and.callFake(async (options) => { + if (options?.callerHoldsOperationLogLock) { + heldLockDrainCount++; + if (heldLockDrainCount === 1) { + throw new Error('transient deferred drain failure'); + } + for (const action of getDeferredActions()) { + acknowledgeDeferredAction(action); + } + } + }); + operationApplierSpy.applyOperations.and.resolveTo({ + appliedOps: [persistedEntry.op], + }); + const dispatchSpy = spyOn( + TestBed.inject(MockStore), + 'dispatch', + ).and.callThrough(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + try { + await service.downloadRemoteOps(mockProvider); + + expect(heldLockDrainCount).toBe(2); + expect(dispatchSpy).toHaveBeenCalledWith( + jasmine.objectContaining({ + type: localAction.type, + meta: jasmine.objectContaining({ isRemote: true }), + }), + ); + expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith( + [persistedEntry.op], + jasmine.objectContaining({ + skipReducerDispatch: true, + skipDeferredLocalActions: true, + remoteApplyWindowAlreadyOpen: true, + }), + ); + expect(mockProvider.setLastServerSeq).toHaveBeenCalledOnceWith(1); + } finally { + clearDeferredActions(); + } + }); + it('should NOT throw LocalDataConflictError on normal incremental sync (no snapshotState)', async () => { // This tests the regression fix: normal incremental syncs should NOT throw // LocalDataConflictError, even if the client has unsynced ops. @@ -2102,10 +3125,10 @@ describe('OperationLogSyncService', () => { await service.downloadRemoteOps(mockProvider); - expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith( - [snapshotIncludedOp], - 'remote', - ); + expect( + syncHydrationServiceSpy.hydrateFromRemoteSync.calls.mostRecent().args[4] + ?.snapshotIncludedOps, + ).toEqual([snapshotIncludedOp]); expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledOnceWith( [postSnapshotOp], {}, @@ -2113,6 +3136,86 @@ describe('OperationLogSyncService', () => { expect(callOrder).toEqual(['processRemoteOps', 'setLastServerSeq']); }); + it('retries only the remaining split suffix after an incompatible op blocks a hydrated batch', async () => { + const op5: Operation = { + id: 'post-snapshot-op-5', + clientId: 'client-B', + actionType: 'test' as ActionType, + opType: OpType.Create, + entityType: 'TASK', + entityId: 'task-5', + payload: {}, + vectorClock: { clientB: 5 }, + timestamp: 5, + schemaVersion: 1, + }; + const op6: Operation = { + ...op5, + id: 'post-snapshot-op-6', + entityId: 'task-6', + vectorClock: { clientB: 6 }, + timestamp: 6, + }; + const snapshotResult = (newOps: Operation[]): DownloadResult => + ({ + newOps, + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['snapshot-task'] } }, + snapshotVectorClock: { clientB: 4 }, + snapshotAppliedOpIds: [], + latestServerSeq: 6, + }) as DownloadResult; + downloadServiceSpy.downloadRemoteOps.and.returnValues( + Promise.resolve(snapshotResult([op5, op6])), + Promise.resolve(snapshotResult([op6])), + ); + opLogStoreSpy.getVectorClock.and.returnValues( + Promise.resolve(null), + Promise.resolve({ clientB: 5 }), + ); + remoteOpsProcessingServiceSpy.processRemoteOps.and.returnValues( + Promise.resolve({ + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, + }), + Promise.resolve({ + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, + }), + ); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + const setLastServerSeqSpy = jasmine + .createSpy('setLastServerSeq') + .and.resolveTo(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: setLastServerSeqSpy, + } as unknown as OperationSyncCapable; + + const first = await service.downloadRemoteOps(mockProvider); + const second = await service.downloadRemoteOps(mockProvider); + + expect(first.kind).toBe('blocked_incompatible'); + expect(second.kind).toBe('ops_processed'); + expect(syncHydrationServiceSpy.hydrateFromRemoteSync).toHaveBeenCalledTimes(1); + expect(remoteOpsProcessingServiceSpy.processRemoteOps.calls.allArgs()).toEqual([ + [[op5, op6], {}], + [[op6], {}], + ]); + expect(setLastServerSeqSpy).toHaveBeenCalledOnceWith(6); + }); + 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')]), @@ -2464,6 +3567,66 @@ describe('OperationLogSyncService', () => { expect(setLastServerSeqSpy).toHaveBeenCalledWith(1); }); + it('should apply a split suffix before advancing when local dominates only the snapshot', async () => { + const suffixOp: Operation = { + id: 'post-snapshot-op', + clientId: 'windowsClient', + actionType: 'test' as ActionType, + opType: OpType.Create, + entityType: 'TASK', + entityId: 'remote-new-task', + payload: {}, + vectorClock: { windowsClient: 2 }, + timestamp: 2, + schemaVersion: 1, + }; + opLogStoreSpy.getVectorClock.and.resolveTo({ + windowsClient: 1, + iosClient: 5, + }); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [suffixOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { tasks: [{ id: 'snapshot-task' }] }, + snapshotVectorClock: { windowsClient: 1 }, + snapshotAppliedOpIds: [], + latestServerSeq: 2, + }); + remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({ + localWinOpsCreated: 2, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, + }); + const syncHydrationServiceSpy = TestBed.inject( + SyncHydrationService, + ) as jasmine.SpyObj; + const setLastServerSeqSpy = jasmine + .createSpy('setLastServerSeq') + .and.resolveTo(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: setLastServerSeqSpy, + } as unknown as OperationSyncCapable; + + const result = await service.downloadRemoteOps(mockProvider); + + expect(result.kind).toBe('ops_processed'); + if (result.kind === 'ops_processed') { + expect(result.localWinOpsCreated).toBe(2); + } + expect(syncHydrationServiceSpy.hydrateFromRemoteSync).not.toHaveBeenCalled(); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledOnceWith( + [suffixOp], + {}, + ); + expect(setLastServerSeqSpy).toHaveBeenCalledOnceWith(2); + }); + it('should NOT persist accompanying newOps on the dominate path — would corrupt per-entity frontiers (codex re-review)', async () => { // VectorClockService.getEntityFrontier() builds per-entity frontiers // by iterating the op log in seq order with last-write-wins semantics. @@ -3622,7 +4785,7 @@ describe('OperationLogSyncService', () => { ); }); - it('should reset lastServerSeq to 0', async () => { + it('should acknowledge only the final cursor after a successful rebuild', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [makeRemoteOp()], needsFullStateUpload: false, @@ -3641,10 +4804,10 @@ describe('OperationLogSyncService', () => { await service.forceDownloadRemoteState(mockProvider); - expect(setLastServerSeqSpy).toHaveBeenCalledWith(0); + expect(setLastServerSeqSpy).toHaveBeenCalledOnceWith(1); }); - it('should reset the external cursor before committing the local replacement', async () => { + it('should commit the local replacement before acknowledging the external cursor', async () => { const callOrder: string[] = []; downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [makeRemoteOp()], @@ -3662,13 +4825,13 @@ describe('OperationLogSyncService', () => { setLastServerSeq: jasmine .createSpy('setLastServerSeq') .and.callFake(async (seq: number) => { - if (seq === 0) callOrder.push('cursor-zero'); + callOrder.push(`cursor-${seq}`); }), } as unknown as OperationSyncCapable; await service.forceDownloadRemoteState(mockProvider); - expect(callOrder.slice(0, 2)).toEqual(['cursor-zero', 'replace']); + expect(callOrder).toEqual(['replace', 'cursor-1']); }); it('should download raw history: forceFromSeq0 AND includeOwnAndAppliedOps', async () => { @@ -3872,7 +5035,7 @@ describe('OperationLogSyncService', () => { ); }); - it('should NOT advance the cursor past 0 when replay blocks on a migration failure', async () => { + it('should not acknowledge the cursor when replay blocks on a migration failure', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [makeRemoteOp()], needsFullStateUpload: false, @@ -3898,8 +5061,7 @@ describe('OperationLogSyncService', () => { await expectAsync( service.forceDownloadRemoteState(mockProvider), ).toBeRejectedWithError(/USE_REMOTE incomplete/); - expect(setLastServerSeqSpy).toHaveBeenCalledWith(0); - expect(setLastServerSeqSpy).not.toHaveBeenCalledWith(50); + expect(setLastServerSeqSpy).not.toHaveBeenCalled(); }); it('should process downloaded ops without confirmation', async () => { @@ -3975,9 +5137,7 @@ describe('OperationLogSyncService', () => { await service.forceDownloadRemoteState(mockProvider); - // First call is reset to 0, second call is update to latestServerSeq - expect(setLastServerSeqSpy).toHaveBeenCalledWith(0); - expect(setLastServerSeqSpy).toHaveBeenCalledWith(50); + expect(setLastServerSeqSpy).toHaveBeenCalledOnceWith(50); }); it('should REJECT an empty remote instead of silently succeeding', async () => { @@ -4063,14 +5223,19 @@ describe('OperationLogSyncService', () => { latestServerSeq: 1, }); + const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); const mockProvider = { supportsOperationSync: true, - setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + setLastServerSeq: setLastServerSeqSpy, } as any; await expectAsync(service.forceDownloadRemoteState(mockProvider)).toBeRejectedWith( error, ); + // A failed replacement must not acknowledge the staged download baseline. + // The raw-rebuild marker drives seq-0 recovery after a committed crash, so + // there is no need for an eager external cursor reset before the transaction. + expect(setLastServerSeqSpy).not.toHaveBeenCalled(); }); // Issue #7330: post-sync validation failure must be surfaced so @@ -4120,6 +5285,43 @@ describe('OperationLogSyncService', () => { }, ); }); + + it('hydrates a split snapshot and replays only its post-snapshot suffix', async () => { + const snapshotOp = makeRemoteOp('snapshot-op'); + const suffixOp = { + ...makeRemoteOp('suffix-op'), + entityId: 'task-after-snapshot', + vectorClock: { remote: 2 }, + }; + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [snapshotOp, suffixOp], + needsFullStateUpload: false, + success: true, + providerMode: 'fileSnapshotOps', + failedFileCount: 0, + snapshotState: { task: { ids: ['task1'] } }, + snapshotVectorClock: { remote: 1 }, + snapshotAppliedOpIds: [snapshotOp.id], + latestServerSeq: 2, + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + await service.forceDownloadRemoteState(mockProvider); + + expect(opLogStoreSpy.appendSnapshotIncludedOps).toHaveBeenCalledOnceWith([ + snapshotOp, + ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledOnceWith( + [suffixOp], + { + skipConflictDetection: true, + callerHoldsOperationLogLock: true, + }, + ); + }); }); describe('_hasMeaningfulStoreData detection for first-time sync', () => { 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 b43b982610..2f402d9569 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -79,6 +79,8 @@ import { 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'; +import { HydrationStateService } from '../apply/hydration-state.service'; +import { getDeferredActions } from '../capture/operation-capture.meta-reducer'; type RemoteOpsProcessingResult = Awaited< ReturnType @@ -179,6 +181,7 @@ export class OperationLogSyncService { private syncImportConflictCoordinator = inject(SyncImportConflictCoordinatorService); private providerManager = inject(SyncProviderManager); private operationApplier = inject(OperationApplierService); + private hydrationState = inject(HydrationStateService); private injector = inject(Injector); /** @@ -610,7 +613,10 @@ export class OperationLogSyncService { `OperationLogSyncService: Local vector clock ${hydrationPlan.comparison} remote snapshot — ` + 'skipping snapshot hydration (local already has all remote data).', ); - // Deliberately do NOT call appendBatchSkipDuplicates(result.newOps). + // Deliberately do NOT append operations already represented by the + // snapshot. Split files can also return a newer suffix, however, and + // snapshot-clock dominance says nothing about that suffix. Apply it + // normally before advancing the cursor. // VectorClockService.getEntityFrontier() builds per-entity frontiers // by iterating the op log in seq order with last-write-wins semantics. // Appending historical remote ops at the current tail would regress @@ -618,21 +624,44 @@ export class OperationLogSyncService { // which then lets future remote ops be classified as non-conflicting // and silently overwrite local changes. // - // The trade-off: those ops keep coming back in result.newOps on each + // The trade-off: snapshot-included ops keep coming back in result.newOps on each // sync until the file's snapshot advances or the user uploads their // own snapshot. They are never re-applied to state, because (a) the // dominate-check skips state mutation, and (b) the regular hydration // path replaces state wholesale from snapshotState, not by replaying // individual ops. So the cost is bounded re-download bandwidth, not // data corruption. + const { postSnapshotOps } = this._partitionSnapshotOps( + result.newOps, + result.snapshotAppliedOpIds, + ); + let suffixProcessResult: RemoteOpsProcessingResult | undefined; + if (postSnapshotOps.length > 0) { + suffixProcessResult = await this._processRemoteOpsWithStartupCleanup( + postSnapshotOps, + undefined, + [], + ); + if (suffixProcessResult.blockedByIncompatibleOp) { + return { kind: 'blocked_incompatible' }; + } + } if (result.latestServerSeq !== undefined) { await syncProvider.setLastServerSeq(result.latestServerSeq); } - return { - kind: 'no_new_ops', - allOpClocks: result.allOpClocks, - snapshotVectorClock: result.snapshotVectorClock, - }; + return suffixProcessResult + ? { + kind: 'ops_processed', + newOpsCount: postSnapshotOps.length, + localWinOpsCreated: suffixProcessResult.localWinOpsCreated, + allOpClocks: result.allOpClocks, + snapshotVectorClock: result.snapshotVectorClock, + } + : { + kind: 'no_new_ops', + allOpClocks: result.allOpClocks, + snapshotVectorClock: result.snapshotVectorClock, + }; } OpLog.normal( @@ -835,14 +864,20 @@ export class OperationLogSyncService { } } - // Hydrate state from snapshot - DON'T create SYNC_IMPORT for file-based bootstrap. - // Creating SYNC_IMPORT would trigger "clean slate" semantics that filter concurrent - // ops from other clients (via SyncImportFilterService). For normal file-based sync, - // we want to merge concurrent work, not discard it. - await this.syncHydrationService.hydrateFromRemoteSync( - result.snapshotState as Record, - result.snapshotVectorClock, - false, // Don't create SYNC_IMPORT for file-based bootstrap + const initialUnsyncedOpIds = new Set(unsyncedOps.map((entry) => entry.op.id)); + + // Single-file snapshots are current through every returned recent op. Split + // snapshots can lag behind sync-ops.json, so only the explicitly-listed + // snapshot ops may be recorded as already applied; the remaining suffix + // must run through normal remote-op processing before the cursor advances. + // An absent list keeps the legacy single-file contract (all ops included). + const { snapshotIncludedOps, postSnapshotOps } = this._partitionSnapshotOps( + result.newOps, + result.snapshotAppliedOpIds, + ); + + await this.writeFlushService.flushThenRunExclusive(() => + this._hydrateSnapshotExclusive(result, initialUnsyncedOpIds, snapshotIncludedOps), ); // Now that the remote snapshot is applied, it's safe to drop the @@ -851,44 +886,14 @@ export class OperationLogSyncService { // so the next attempt can retry. await this._discardStartupOps(startupOpIdsToDiscard); - // Single-file snapshots are current through every returned recent op. Split - // snapshots can lag behind sync-ops.json, so only the explicitly-listed - // snapshot ops may be recorded as already applied; the remaining suffix - // must run through normal remote-op processing before the cursor advances. - // An absent list keeps the legacy single-file contract (all ops included). - const snapshotAppliedOpIds = result.snapshotAppliedOpIds - ? new Set(result.snapshotAppliedOpIds) - : null; - const snapshotIncludedOps = snapshotAppliedOpIds - ? result.newOps.filter((op) => snapshotAppliedOpIds.has(op.id)) - : result.newOps; - const postSnapshotOps = snapshotAppliedOpIds - ? result.newOps.filter((op) => !snapshotAppliedOpIds.has(op.id)) - : []; - - // Record operations already represented by the snapshot so future whole- - // file downloads deduplicate them without replaying their effects twice. - if (snapshotIncludedOps.length > 0) { - const appendResult = await this.opLogStore.appendBatchSkipDuplicates( - snapshotIncludedOps, - 'remote', - ); - OpLog.normal( - `OperationLogSyncService: Wrote ${appendResult.writtenOps.length} snapshot ops to IndexedDB ` + - '(prevents duplication on next sync cycle).' + - (appendResult.skippedCount > 0 - ? ` Skipped ${appendResult.skippedCount} duplicate(s).` - : ''), - ); - } - + let suffixProcessResult: RemoteOpsProcessingResult | undefined; if (postSnapshotOps.length > 0) { - const processResult = await this._processRemoteOpsWithStartupCleanup( + suffixProcessResult = await this._processRemoteOpsWithStartupCleanup( postSnapshotOps, undefined, [], ); - if (processResult.blockedByIncompatibleOp) { + if (suffixProcessResult.blockedByIncompatibleOp) { return { kind: 'blocked_incompatible' }; } } @@ -900,11 +905,19 @@ export class OperationLogSyncService { OpLog.normal('OperationLogSyncService: Snapshot hydration complete.'); - return { - kind: 'snapshot_hydrated', - allOpClocks: result.allOpClocks, - snapshotVectorClock: result.snapshotVectorClock, - }; + return suffixProcessResult + ? { + kind: 'ops_processed', + newOpsCount: postSnapshotOps.length, + localWinOpsCreated: suffixProcessResult.localWinOpsCreated, + allOpClocks: result.allOpClocks, + snapshotVectorClock: result.snapshotVectorClock, + } + : { + kind: 'snapshot_hydrated', + allOpClocks: result.allOpClocks, + snapshotVectorClock: result.snapshotVectorClock, + }; } if (result.newOps.length === 0) { @@ -1249,6 +1262,188 @@ export class OperationLogSyncService { } } + private _partitionSnapshotOps( + ops: Operation[], + snapshotAppliedOpIds: string[] | undefined, + ): { snapshotIncludedOps: Operation[]; postSnapshotOps: Operation[] } { + if (snapshotAppliedOpIds === undefined) { + return { snapshotIncludedOps: ops, postSnapshotOps: [] }; + } + const includedIds = new Set(snapshotAppliedOpIds); + return { + snapshotIncludedOps: ops.filter((op) => includedIds.has(op.id)), + postSnapshotOps: ops.filter((op) => !includedIds.has(op.id)), + }; + } + + /** + * Body of the exclusive file-snapshot hydration section: hydrates the remote + * snapshot while user actions are deferred, persists the buffered intents on + * top of it, and replays their archive side effects. Runs inside + * `writeFlushService.flushThenRunExclusive` — no other op-log writer can + * interleave. + */ + private async _hydrateSnapshotExclusive( + result: { + snapshotState?: unknown; + snapshotVectorClock?: Record; + remoteLastModified?: number; + }, + initialUnsyncedOpIds: Set, + snapshotIncludedOps: Operation[], + ): Promise { + let deferredActionsOverwrittenBySnapshot: ReturnType = []; + let didReplaceArchive = false; + let didCommitStateLoad = false; + let hydrationFailed = false; + let hydrationError: unknown; + let isRemoteApplyWindowOpen = true; + // Keep capture in deferred mode from the last pre-hydration durability + // check through the snapshot dispatch. Otherwise an action can be + // persisted against the old state and then silently overwritten by + // loadAllData while this async hydration is in progress. + this.hydrationState.startApplyingRemoteOps(); + try { + try { + const pendingAtHydrationCutoff = await this.opLogStore.getUnsynced(); + const lateDurableOps = pendingAtHydrationCutoff.filter( + (entry) => !initialUnsyncedOpIds.has(entry.op.id), + ); + if (lateDurableOps.length > 0) { + const lastSyncedVectorClock = + (await this.vectorClockService.getSnapshotVectorClock()) ?? null; + throw new LocalDataConflictError( + pendingAtHydrationCutoff.length, + result.snapshotState as Record, + result.snapshotVectorClock, + lastSyncedVectorClock, + result.remoteLastModified, + ); + } + + // Hydrate state from snapshot - DON'T create SYNC_IMPORT for file-based + // bootstrap. Creating it would trigger clean-slate filtering of concurrent + // ops from other clients. + await this.syncHydrationService.hydrateFromRemoteSync( + result.snapshotState as Record, + result.snapshotVectorClock, + false, + undefined, + { + snapshotIncludedOps, + // Capture only actions that ran on the old state. Actions emitted + // by loadAllData effects run after the snapshot reducer and are + // already valid on top of the new state, so replaying those would + // double-apply additive reducers. + beforeStateLoad: () => { + deferredActionsOverwrittenBySnapshot = getDeferredActions(); + }, + afterStateLoad: () => { + didCommitStateLoad = true; + }, + afterArchiveReplacement: () => { + didReplaceArchive = true; + }, + }, + ); + } catch (error) { + hydrationFailed = true; + hydrationError = error; + } + + // The file snapshot baseline transaction either committed state, clock, + // archives, and included remote ops together or left the old baseline + // intact. Deferred intents therefore always drain against a complete + // frontier, including when hydration failed before live-state dispatch. + await this._processSnapshotDeferredActionsWithRetry(); + + if (didCommitStateLoad) { + // Persistent actions dispatched before loadAllData already ran once on + // the old NgRx state. Re-dispatch remote-marked clones synchronously on + // top of the snapshot without capturing a second operation. + for (const action of deferredActionsOverwrittenBySnapshot) { + this.store.dispatch({ + ...action, + meta: { + ...action.meta, + isRemote: true, + }, + }); + } + } + + if (didReplaceArchive) { + // Hydration also replaces archive stores. Re-run archive side effects + // for every local intent created since the cutoff, in durable seq order. + // Keep looping until both the deferred queue and durable archive work + // are empty. The final queue check and remote-window close are + // synchronous, so an action cannot land in the handoff gap. + const restoredLocalOpIds = new Set(initialUnsyncedOpIds); + while (true) { + while (getDeferredActions().length > 0) { + await this._processSnapshotDeferredActionsWithRetry(); + } + + const localOpsNeedingArchiveRestore = (await this.opLogStore.getUnsynced()) + .filter((entry) => !restoredLocalOpIds.has(entry.op.id)) + .sort((a, b) => a.seq - b.seq) + .map((entry) => entry.op); + if (localOpsNeedingArchiveRestore.length === 0) { + if (getDeferredActions().length > 0) continue; + this.hydrationState.endApplyingRemoteOps(); + isRemoteApplyWindowOpen = false; + break; + } + + const applyResult = await this.operationApplier.applyOperations( + localOpsNeedingArchiveRestore, + { + isLocalHydration: false, + skipDeferredLocalActions: true, + skipReducerDispatch: true, + remoteApplyWindowAlreadyOpen: true, + }, + ); + if ( + applyResult.failedOp || + applyResult.appliedOps.length !== localOpsNeedingArchiveRestore.length + ) { + throw new Error( + 'Snapshot hydration incomplete: local archive changes could not be restored.', + ); + } + for (const op of localOpsNeedingArchiveRestore) { + restoredLocalOpIds.add(op.id); + } + } + } else { + while (getDeferredActions().length > 0) { + await this._processSnapshotDeferredActionsWithRetry(); + } + this.hydrationState.endApplyingRemoteOps(); + isRemoteApplyWindowOpen = false; + } + + if (hydrationFailed) throw hydrationError; + } finally { + if (isRemoteApplyWindowOpen) { + this.hydrationState.endApplyingRemoteOps(); + } + } + } + + private async _processSnapshotDeferredActionsWithRetry(): Promise { + try { + await processDeferredActions(this.injector, true); + } catch (error) { + OpLog.warn( + 'OperationLogSyncService: deferred snapshot action drain failed; retrying once', + { name: (error as Error | undefined)?.name }, + ); + await processDeferredActions(this.injector, true); + } + } + private async _discardStartupOpsIfFullStateCommitted( fullStateOpId: string | undefined, startupOpIds: string[], @@ -1521,7 +1716,24 @@ export class OperationLogSyncService { throw new Error('USE_REMOTE aborted: remote returned no data to rebuild from.'); } - const migratedRemoteOps = this._preflightRemoteOperations(result.newOps); + let migratedRemoteOps: Operation[]; + let migratedSnapshotIncludedOps: Operation[] = []; + let migratedPostSnapshotOps: Operation[] = []; + if (hasSnapshotState && result.providerMode === 'fileSnapshotOps') { + const snapshotOps = this._partitionSnapshotOps( + result.newOps, + result.snapshotAppliedOpIds, + ); + migratedSnapshotIncludedOps = this._preflightRemoteOperations( + snapshotOps.snapshotIncludedOps, + ); + migratedPostSnapshotOps = this._preflightRemoteOperations( + snapshotOps.postSnapshotOps, + ); + migratedRemoteOps = [...migratedSnapshotIncludedOps, ...migratedPostSnapshotOps]; + } else { + migratedRemoteOps = this._preflightRemoteOperations(result.newOps); + } const currentSyncConfig = await firstValueFrom(this.store.select(selectSyncConfig)); const localOnlySyncSettings: LocalOnlySyncSettings = { @@ -1666,11 +1878,11 @@ export class OperationLogSyncService { } // The provider cursor lives outside SUP_OPS, so it cannot join the IDB - // transaction. Reset it first: a crash/failure before the transaction - // merely causes a safe re-download onto intact local state, while a - // commit can never become visible with a stale cursor that skips the - // remote history required to rebuild the baseline. - await syncProvider.setLastServerSeq(0); + // transaction. Do not reset it eagerly: for file adapters that call is + // also the durable-apply acknowledgement for the staged download. The + // transaction stores a raw-rebuild-incomplete marker atomically with + // the replacement, and crash recovery always re-downloads from seq 0; + // the cursor is therefore advanced only after replay/hydration succeeds. await this.opLogStore.runRemoteStateReplacement({ baselineState, vectorClock: rebuiltClock, @@ -1707,13 +1919,11 @@ export class OperationLogSyncService { false, // Don't create SYNC_IMPORT ); - // CRITICAL FIX: Write recentOps to IndexedDB after snapshot hydration. - // Same rationale as downloadRemoteOps: file-based providers return ALL - // recentOps on every download and rely on getAppliedOpIds() to filter them. - if (migratedRemoteOps.length > 0) { - const appendResult = await this.opLogStore.appendBatchSkipDuplicates( - migratedRemoteOps, - 'remote', + // Record only operations already represented by the snapshot. A + // split-file suffix must still be dispatched on top of that state. + if (migratedSnapshotIncludedOps.length > 0) { + const appendResult = await this.opLogStore.appendSnapshotIncludedOps( + migratedSnapshotIncludedOps, ); OpLog.normal( `OperationLogSyncService: Wrote ${appendResult.writtenOps.length} snapshot ops to IndexedDB ` + @@ -1724,6 +1934,22 @@ export class OperationLogSyncService { ); } + if (migratedPostSnapshotOps.length > 0) { + const processResult = + await this.remoteOpsProcessingService.processRemoteOps( + migratedPostSnapshotOps, + { + skipConflictDetection: true, + callerHoldsOperationLogLock: true, + }, + ); + if (processResult.blockedByIncompatibleOp) { + throw new Error( + 'USE_REMOTE incomplete: a post-snapshot op failed schema migration during replay.', + ); + } + } + await this._restorePreservedLocalOps(preservedLocalOps); await this._assertNoCaptureRacedWithRebuild(); diff --git a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts index 2b921602a2..c363f76661 100644 --- a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts +++ b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts @@ -188,10 +188,16 @@ describe('Post-sync validation latch (#7330) — integration', () => { 'setVectorClock', 'append', 'loadStateCache', + 'commitFileSnapshotBaseline', ]); opLogStoreHydrationSpy.getLastSeq.and.resolveTo(0); opLogStoreHydrationSpy.getUnsynced.and.resolveTo([]); opLogStoreHydrationSpy.loadStateCache.and.resolveTo(null); + opLogStoreHydrationSpy.commitFileSnapshotBaseline.and.resolveTo({ + seqs: [], + writtenOps: [], + skippedCount: 0, + }); const stateSnapshotSpy = jasmine.createSpyObj('StateSnapshotService', [ 'getStateSnapshot', 'getAllSyncModelDataFromStoreAsync',