mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
fix(sync): preserve local edits during snapshot hydration (#9010)
* 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
This commit is contained in:
Johannes Millan
GitHub
parent
631e6e9710
commit
29df7e7719
7 changed files with 2255 additions and 138 deletions
|
|
@ -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<unknown>;
|
||||
}
|
||||
).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<typeof tx.put>[0],
|
||||
...args: unknown[]
|
||||
): Promise<unknown> => {
|
||||
if (storeName === STORE_NAMES.VECTOR_CLOCK) {
|
||||
throw new Error('injected vector-clock write failure');
|
||||
}
|
||||
return (target.put as (...putArgs: unknown[]) => Promise<unknown>).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', () => {
|
||||
|
|
|
|||
|
|
@ -783,6 +783,151 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
ops: Operation[],
|
||||
source: 'local' | 'remote' = 'local',
|
||||
options?: { pendingApply?: boolean },
|
||||
): Promise<{ seqs: number[]; writtenOps: Operation[]; skippedCount: number }> {
|
||||
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<StoredOperationLogEntry>(
|
||||
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<O
|
|||
let skippedCount = 0;
|
||||
|
||||
const storeNames: OpLogStoreName[] = [STORE_NAMES.OPS];
|
||||
if (advanceSnapshotFrontier) {
|
||||
storeNames.push(STORE_NAMES.STATE_CACHE);
|
||||
}
|
||||
if (ops.some((op) => 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<StateCacheEntry>(
|
||||
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<StoredOperationLogEntry>(
|
||||
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<O
|
|||
op.id,
|
||||
);
|
||||
if (existingKey !== undefined) {
|
||||
if (typeof existingKey !== 'number') {
|
||||
throw new Error('Operation sequence key is not numeric');
|
||||
}
|
||||
lastIncludedSeq = Math.max(lastIncludedSeq, existingKey);
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
|
|
@ -815,9 +999,17 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
const entry = this._buildStoredEntry(op, source, options);
|
||||
const seq = await tx.add(STORE_NAMES.OPS, entry);
|
||||
await this._recordFullStateOpInTx(tx, entry.op, seq);
|
||||
lastIncludedSeq = Math.max(lastIncludedSeq, seq);
|
||||
seqs.push(seq);
|
||||
writtenOps.push(op);
|
||||
}
|
||||
|
||||
if (stateCache) {
|
||||
await tx.put(STORE_NAMES.STATE_CACHE, {
|
||||
...stateCache,
|
||||
lastAppliedOpSeq: Math.max(stateCache.lastAppliedOpSeq, lastIncludedSeq),
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
if (skippedCount > 0) {
|
||||
|
|
|
|||
|
|
@ -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<VectorClockService>;
|
||||
let mockValidateStateService: jasmine.SpyObj<ValidateStateService>;
|
||||
let mockSnackService: jasmine.SpyObj<SnackService>;
|
||||
let mockArchiveDbAdapter: jasmine.SpyObj<ArchiveDbAdapter>;
|
||||
let mockLockService: jasmine.SpyObj<LockService>;
|
||||
|
||||
// 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,
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>,
|
||||
remoteVectorClock?: Record<string, number>,
|
||||
createSyncImportOp: boolean = true,
|
||||
syncImportReason?: SyncImportReason,
|
||||
hooks?: SnapshotHydrationHooks,
|
||||
): Promise<void> {
|
||||
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<string, unknown>;
|
||||
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<string, unknown>
|
||||
| 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);
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -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<RemoteOpsProcessingService['processRemoteOps']>
|
||||
|
|
@ -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<string, unknown>,
|
||||
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<string, number>;
|
||||
remoteLastModified?: number;
|
||||
},
|
||||
initialUnsyncedOpIds: Set<string>,
|
||||
snapshotIncludedOps: Operation[],
|
||||
): Promise<void> {
|
||||
let deferredActionsOverwrittenBySnapshot: ReturnType<typeof getDeferredActions> = [];
|
||||
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<string, unknown>,
|
||||
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<string, unknown>,
|
||||
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<void> {
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue