From 29df7e7719adb661a702d17b6ea3d48fc96f355d Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Tue, 14 Jul 2026 21:53:05 +0200 Subject: [PATCH] fix(sync): preserve local edits during snapshot hydration (#9010) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * docs(plugins): add microsoft 365 calendar provider plan * docs(plugins): add microsoft 365 calendar provider plan * docs(ios): plan internal testflight builds * fix(sync): prevent file-based sync data loss Preserve post-snapshot operations and stage remote baselines until durable apply. Use conditional writes and a resumable marker for provider concurrency and legacy migration. Report trustworthy remote timestamps and document transport limits. Refs #8960 * fix(sync): avoid biased conflict recommendations Highlight a side only when its timestamp or known change count is strictly greater, leaving unknown and tied metadata neutral. * fix(sync): preserve local edits during snapshot hydration Extracts the one reachable-data-loss fix from the #9005 hardening branch (feat/sync-recovery-hardening). On the default single-file path, a local action dispatched while an async snapshot hydrate is in flight was persisted against the old state and then clobbered by loadAllData (or permanently dropped by markRejected) — a silent edit loss multi-device users can hit on gap-detected re-sync. Runs hydration inside writeFlushService.flushThenRunExclusive so capture stays in deferred mode across the snapshot dispatch, commits the snapshot baseline atomically (commitFileSnapshotBaseline), then replays the buffered local intents and their archive side effects onto the new baseline. Deliberately excludes #9005's snapshot two-phase-commit, structural validation, split-migration crash-resume, and gap persistence: reachability analysis found those defend non-reachable (structural validation), already-fail-safe (crash-mid-write), or default-off (split sync) failure modes, at the cost of ~1k lines and two permanent on-disk formats. * test(sync): add commitFileSnapshotBaseline to hydration spy --- .../operation-log-store.service.spec.ts | 161 +++ .../operation-log-store.service.ts | 192 +++ .../sync-hydration.service.spec.ts | 281 +++- .../persistence/sync-hydration.service.ts | 159 ++- .../sync/operation-log-sync.service.spec.ts | 1236 ++++++++++++++++- .../op-log/sync/operation-log-sync.service.ts | 358 ++++- .../post-sync-validation.integration.spec.ts | 6 + 7 files changed, 2255 insertions(+), 138 deletions(-) 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',