From eecf33a4e88b8cab613fa62b1d7eba33847910b7 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Fri, 10 Jul 2026 16:38:16 +0200 Subject: [PATCH] fix(sync): remediate multi-agent review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - USE_REMOTE: fix guaranteed self-deadlock — flushPendingWrites() was called while already holding the non-reentrant sp_op_log lock its own Phase 2 re-acquires, so every "Use Server Data" hung 30s and aborted. New OperationWriteFlushService.flushThenRunExclusive owns the bounded flush->lock->recheck barrier; ServerMigrationService reuses it, which also bounds its previously unbounded snapshot-cutoff retry loop. - USE_REMOTE: make the rebuild crash-resumable. A raw_rebuild_incomplete META marker is set atomically with the baseline replacement and cleared after the replay commits; the next sync redoes the raw rebuild instead of the normal download (which excludes own ops server-side and would silently lose the un-replayed suffix). The resume keeps the first attempt's pre-replace backup instead of overwriting the single slot with the partial baseline. - Deferred actions: serialize overlapping drains (two concurrent drains could mint two ops for one user intent), stop the drain at the first transient failure instead of persisting successors out of order (the older edit would win LWW everywhere via clock inversion), abandon permanently invalid actions after one attempt (typed PermanentDeferredWriteError) instead of retrying + snacking on every sync forever, and latch the buffer-size warnings to once per stuck window (previously a blocking dev dialog per action past 100); drop the no-op 5000 "hard cap" tier. - writeOperation: post-append bookkeeping failures no longer propagate into the deferred retry loop, which re-appended the same action under a fresh op id (double-apply of additive payloads). - remote-apply: full-state cleanup retains every full-state op of the current batch, so an archive_pending import can no longer be deleted before its archive retry (startup replay would lose a change already visible at runtime). - Hydration: sanitize a malformed stored schemaVersion per-op instead of failing the whole boot into recovery; strict parsing (floor now 1, matching the server contract) stays on the receive/upload paths. - Server: request fingerprints are computed lazily — only when a dedup entry exists, and otherwise after the rate-limit/pre-quota gates but before uploadOps mutates the parsed ops. Fixes the pre-gate hashing of full payloads (authenticated DoS-cost regression) and repairs three sync-fixes retry tests to model true identical-body retries. - Delete dead code: failed-op-ids util (+spec) and MAX_VERSION_SKIP (the removed forward-compat band); fix the stale clock-merge parity comment. sync-core 210/210, shared-schema 50/50, super-sync-server 825/825, and all touched Angular specs pass. --- packages/shared-schema/src/index.ts | 6 +- packages/shared-schema/src/schema-version.ts | 9 +- .../services/request-deduplication.service.ts | 19 +- .../src/sync/sync.routes.ops-handler.ts | 29 +- .../src/sync/sync.routes.snapshot-handler.ts | 44 ++- .../src/sync/sync.service.ts | 8 +- .../request-deduplication.service.spec.ts | 27 +- .../tests/sync-compressed-body.routes.spec.ts | 2 +- .../tests/sync-fixes.spec.ts | 17 +- packages/sync-core/src/remote-apply.ts | 15 +- packages/sync-core/tests/remote-apply.spec.ts | 32 ++ .../op-log/apply/failed-op-ids.util.spec.ts | 27 -- src/app/op-log/apply/failed-op-ids.util.ts | 25 -- .../operation-capture.meta-reducer.spec.ts | 43 ++- .../capture/operation-capture.meta-reducer.ts | 43 +-- .../capture/operation-log.effects.spec.ts | 98 +++++- .../op-log/capture/operation-log.effects.ts | 156 +++++++--- src/app/op-log/core/errors/sync-errors.ts | 10 + src/app/op-log/persistence/db-keys.const.ts | 9 + .../operation-log-hydrator.service.spec.ts | 39 ++- .../operation-log-hydrator.service.ts | 37 ++- .../operation-log-store.service.spec.ts | 20 ++ .../operation-log-store.service.ts | 30 ++ .../persistence/schema-migration.service.ts | 10 +- .../sync/operation-log-sync.service.spec.ts | 122 ++++++++ .../op-log/sync/operation-log-sync.service.ts | 279 ++++++++++-------- .../operation-write-flush.service.spec.ts | 72 +++++ .../sync/operation-write-flush.service.ts | 47 +++ .../sync/server-migration.service.spec.ts | 21 +- .../op-log/sync/server-migration.service.ts | 219 +++++++------- .../schema-version-sync.integration.spec.ts | 6 - 31 files changed, 1069 insertions(+), 452 deletions(-) delete mode 100644 src/app/op-log/apply/failed-op-ids.util.spec.ts delete mode 100644 src/app/op-log/apply/failed-op-ids.util.ts diff --git a/packages/shared-schema/src/index.ts b/packages/shared-schema/src/index.ts index 27a706441a..ee17caaf40 100644 --- a/packages/shared-schema/src/index.ts +++ b/packages/shared-schema/src/index.ts @@ -1,9 +1,5 @@ // Schema version constants -export { - CURRENT_SCHEMA_VERSION, - MIN_SUPPORTED_SCHEMA_VERSION, - MAX_VERSION_SKIP, -} from './schema-version'; +export { CURRENT_SCHEMA_VERSION, MIN_SUPPORTED_SCHEMA_VERSION } from './schema-version'; // Types export type { diff --git a/packages/shared-schema/src/schema-version.ts b/packages/shared-schema/src/schema-version.ts index 92997bb47d..b98ad6097d 100644 --- a/packages/shared-schema/src/schema-version.ts +++ b/packages/shared-schema/src/schema-version.ts @@ -10,9 +10,6 @@ export const CURRENT_SCHEMA_VERSION = 2; */ export const MIN_SUPPORTED_SCHEMA_VERSION = 1; -/** - * Maximum version difference we tolerate before forcing an app update. - * If remote data is more than MAX_VERSION_SKIP versions ahead, - * the user must update their app. - */ -export const MAX_VERSION_SKIP = 3; +// NOTE: there is deliberately NO forward-compat band: any op from a NEWER +// schema version is blocked outright (the client cannot know how to interpret +// it), and the user is prompted to update the app. diff --git a/packages/super-sync-server/src/sync/services/request-deduplication.service.ts b/packages/super-sync-server/src/sync/services/request-deduplication.service.ts index 2ea3b1e496..b7fc4e2142 100644 --- a/packages/super-sync-server/src/sync/services/request-deduplication.service.ts +++ b/packages/super-sync-server/src/sync/services/request-deduplication.service.ts @@ -63,13 +63,19 @@ export class RequestDeduplicationService { /** * Check if a request has already been processed for this user + namespace. - * @returns Cached results if found and not expired, null otherwise. + * + * @param getFingerprint lazy fingerprint supplier — hashing the full request + * body is expensive (stable stringify + SHA-256 over up to multi-MB + * payloads), so it must only run when an entry for this requestId actually + * exists (genuine retries are rare) and never for first-time requests. + * @returns Cached results if found, not expired, and fingerprint-matching; + * null otherwise. */ checkDeduplication( userId: number, namespace: N, requestId: string, - fingerprint?: string, + getFingerprint?: () => string, ): DedupPayload | null { const key = this._key(userId, namespace, requestId); const entry = this.cache.get(key); @@ -80,7 +86,14 @@ export class RequestDeduplicationService { } // Defensive: keying already isolates namespaces, but verify before casting. if (entry.namespace !== namespace) return null; - if (fingerprint !== undefined && entry.fingerprint !== fingerprint) return null; + if (getFingerprint !== undefined) { + // A pre-fingerprint (legacy) entry cannot prove body equality — treat as + // a miss so the request is re-processed (safe: the server dedups ops by + // id anyway). + if (entry.fingerprint === undefined || entry.fingerprint !== getFingerprint()) { + return null; + } + } return entry.results as DedupPayload; } diff --git a/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts b/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts index 24ed6eed35..56b260cdc5 100644 --- a/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts +++ b/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts @@ -87,9 +87,15 @@ export const uploadOpsHandler = async ( } const { ops, clientId, lastKnownServerSeq, requestId } = parseResult.data; - const requestFingerprint = requestId - ? createOpsRequestFingerprint(clientId, ops as unknown as Operation[]) - : undefined; + // Lazy + memoized: hashing the full ops payload is expensive and must not + // run before the rate-limit gate below, nor on first-time requests — the + // dedup check only invokes it when an entry for this requestId exists. + let memoizedFingerprint: string | undefined; + const getRequestFingerprint = (): string => + (memoizedFingerprint ??= createOpsRequestFingerprint( + clientId, + ops as unknown as Operation[], + )); const syncService = getSyncService(); Logger.info( @@ -119,7 +125,7 @@ export const uploadOpsHandler = async ( const cachedResults = syncService.checkOpsRequestDedup( userId, requestId, - requestFingerprint, + getRequestFingerprint, ); if (cachedResults) { Logger.info(`[user:${userId}] Returning cached results for request ${requestId}`); @@ -170,6 +176,14 @@ export const uploadOpsHandler = async ( } } + // Pin the fingerprint BEFORE processing: uploadOps mutates the parsed ops + // (e.g. vector-clock pruning), so hashing after it would never match the + // retry's pre-processing hash. Still after the rate-limit gate above, so a + // rate-limited client cannot burn CPU on it. + if (requestId) { + getRequestFingerprint(); + } + const results = await syncService.runWithStorageUsageLock( userId, async () => { @@ -222,7 +236,12 @@ export const uploadOpsHandler = async ( requestId && !results.some((r) => r.errorCode === SYNC_ERROR_CODES.INTERNAL_ERROR) ) { - syncService.cacheOpsRequestResults(userId, requestId, results, requestFingerprint); + syncService.cacheOpsRequestResults( + userId, + requestId, + results, + getRequestFingerprint(), + ); } const accepted = results.filter((r) => r.accepted).length; diff --git a/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts b/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts index 8f59879597..281e695406 100644 --- a/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts +++ b/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts @@ -100,26 +100,31 @@ export const uploadSnapshotHandler = async ( requestId, } = snapshotRequest; const syncService = getSyncService(); - const requestFingerprint = requestId - ? createSnapshotRequestFingerprint({ - state, - clientId, - reason, - vectorClock, - schemaVersion, - isPayloadEncrypted, - syncImportReason, - opId, - isCleanSlate, - snapshotOpType, - }) - : undefined; + // Lazy + memoized: fingerprinting stable-stringifies + SHA-256-hashes the + // full (possibly multi-MB plaintext) snapshot state. It must not run + // before the pre-quota gate below, and first-time requests never need it — + // the dedup check only invokes it when an entry for this requestId exists + // (a genuine retry). + let memoizedFingerprint: string | undefined; + const getRequestFingerprint = (): string => + (memoizedFingerprint ??= createSnapshotRequestFingerprint({ + state, + clientId, + reason, + vectorClock, + schemaVersion, + isPayloadEncrypted, + syncImportReason, + opId, + isCleanSlate, + snapshotOpType, + })); if (requestId) { const cachedResponse = syncService.checkSnapshotRequestDedup( userId, requestId, - requestFingerprint, + getRequestFingerprint, ); if (cachedResponse) { Logger.info( @@ -156,6 +161,13 @@ export const uploadSnapshotHandler = async ( } } + // Pin the fingerprint BEFORE processing mutates anything, but AFTER the + // pre-quota gate above so quota-exhausted clients cannot burn CPU on the + // full-state hash. + if (requestId) { + getRequestFingerprint(); + } + // Reject duplicate SYNC_IMPORT before we acquire the per-user lock — a // duplicate rejection is cheap (one indexed lookup) and skipping the // lock lets concurrent legitimate clients keep moving. @@ -363,7 +375,7 @@ export const uploadSnapshotHandler = async ( userId, requestId, responseBody, - requestFingerprint, + getRequestFingerprint(), ); } diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index d4ac70c58d..113781f822 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -457,13 +457,13 @@ export class SyncService { checkOpsRequestDedup( userId: number, requestId: string, - fingerprint?: string, + getFingerprint?: () => string, ): UploadResult[] | null { return this.requestDeduplicationService.checkDeduplication( userId, 'ops', requestId, - fingerprint, + getFingerprint, ); } @@ -485,13 +485,13 @@ export class SyncService { checkSnapshotRequestDedup( userId: number, requestId: string, - fingerprint?: string, + getFingerprint?: () => string, ): SnapshotDedupResponse | null { return this.requestDeduplicationService.checkDeduplication( userId, 'snapshot', requestId, - fingerprint, + getFingerprint, ); } diff --git a/packages/super-sync-server/tests/request-deduplication.service.spec.ts b/packages/super-sync-server/tests/request-deduplication.service.spec.ts index 41acd782df..d64ca375a5 100644 --- a/packages/super-sync-server/tests/request-deduplication.service.spec.ts +++ b/packages/super-sync-server/tests/request-deduplication.service.spec.ts @@ -44,13 +44,36 @@ describe('RequestDeduplicationService', () => { service.cacheResults(1, 'ops', 'request-123', mockResults, 'fingerprint-a'); expect( - service.checkDeduplication(1, 'ops', 'request-123', 'fingerprint-a'), + service.checkDeduplication(1, 'ops', 'request-123', () => 'fingerprint-a'), ).toEqual(mockResults); expect( - service.checkDeduplication(1, 'ops', 'request-123', 'fingerprint-b'), + service.checkDeduplication(1, 'ops', 'request-123', () => 'fingerprint-b'), ).toBeNull(); }); + it('should not compute the fingerprint when no entry exists for the requestId', () => { + // The fingerprint hashes the full request body — first-time requests + // (the overwhelming majority) must not pay that cost. + const getFingerprint = vi.fn(() => 'fingerprint-a'); + + expect(service.checkDeduplication(1, 'ops', 'request-999', getFingerprint)).toBe( + null, + ); + expect(getFingerprint).not.toHaveBeenCalled(); + }); + + it('should treat a legacy entry without fingerprint as a miss when a fingerprint is supplied', () => { + const mockResults = createMockResults(1); + service.cacheResults(1, 'ops', 'request-123', mockResults); + + const getFingerprint = vi.fn(() => 'fingerprint-a'); + expect( + service.checkDeduplication(1, 'ops', 'request-123', getFingerprint), + ).toBeNull(); + // No entry fingerprint to compare against — the hash is never computed. + expect(getFingerprint).not.toHaveBeenCalled(); + }); + it('should return null for expired request', () => { const mockResults = createMockResults(1); service.cacheResults(1, 'ops', 'request-123', mockResults); diff --git a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts index 60d282be76..0acbb475f0 100644 --- a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts +++ b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts @@ -810,7 +810,7 @@ describe('Sync compressed body routes', () => { expect(mocks.syncService.checkSnapshotRequestDedup).toHaveBeenCalledWith( 1, 'snapshot-v1-retry', - expect.any(String), + expect.any(Function), ); expect(mocks.syncService.checkOpsRequestDedup).not.toHaveBeenCalled(); expect(mocks.prisma.operation.findFirst).not.toHaveBeenCalled(); diff --git a/packages/super-sync-server/tests/sync-fixes.spec.ts b/packages/super-sync-server/tests/sync-fixes.spec.ts index a0a163bddd..96a1a65469 100644 --- a/packages/super-sync-server/tests/sync-fixes.spec.ts +++ b/packages/super-sync-server/tests/sync-fixes.spec.ts @@ -303,6 +303,9 @@ describe('Sync System Fixes', () => { const clientA = 'client-a'; const clientB = 'client-b'; const requestId = 'req-123'; + // A true retry resends the IDENTICAL body — the request fingerprint + // treats a same-requestId/different-body request as a cache miss. + const opA = createOp(clientA, { entityId: 'task-a' }); // Step 1: Client A uploads with requestId const firstResponse = await app.inject({ @@ -310,7 +313,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientA, { entityId: 'task-a' })], + ops: [opA], clientId: clientA, lastKnownServerSeq: 0, requestId, @@ -339,7 +342,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientA, { entityId: 'task-a' })], + ops: [opA], clientId: clientA, lastKnownServerSeq: clientASeq, requestId, @@ -366,6 +369,7 @@ describe('Sync System Fixes', () => { const clientA = 'client-a'; const clientB = 'client-b'; const requestId = 'req-456'; + const opA = createOp(clientA, { entityId: 'task-1' }); // Client A uploads await app.inject({ @@ -373,7 +377,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientA, { entityId: 'task-1' })], + ops: [opA], clientId: clientA, lastKnownServerSeq: 0, requestId, @@ -397,7 +401,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientA, { entityId: 'task-1' })], + ops: [opA], clientId: clientA, lastKnownServerSeq: 0, requestId, @@ -488,6 +492,7 @@ describe('Sync System Fixes', () => { it('should return cached results even when quota would be exceeded on retry', async () => { const clientId = uuidv7(); const requestId = uuidv7(); + const op = createOp(clientId, { entityId: 'task-1' }); // First request with requestId const firstResponse = await app.inject({ @@ -495,7 +500,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientId, { entityId: 'task-1' })], + ops: [op], clientId, lastKnownServerSeq: 0, requestId, @@ -513,7 +518,7 @@ describe('Sync System Fixes', () => { url: '/api/sync/ops', headers: { authorization: `Bearer ${authToken}` }, payload: { - ops: [createOp(clientId, { entityId: 'task-1' })], + ops: [op], clientId, lastKnownServerSeq: 0, requestId, diff --git a/packages/sync-core/src/remote-apply.ts b/packages/sync-core/src/remote-apply.ts index 3cc8f32f77..9676c3a311 100644 --- a/packages/sync-core/src/remote-apply.ts +++ b/packages/sync-core/src/remote-apply.ts @@ -74,7 +74,8 @@ const emptyRemoteApplyResult = < * 2. apply only the newly appended ops through the host applier; * 3. mark successfully applied seqs; * 4. merge applied remote vector clocks; - * 5. retain only the newest applied full-state ops when configured; + * 5. after applying a full-state op, clear older full-state ops while + * retaining every full-state op of this batch (incl. archive_pending); * 6. mark the failed op and remaining ops as failed on partial error — their * reducer effects committed with the bulk dispatch; only their archive side * effects are outstanding (see ApplyOperationsResult.failedOp). @@ -145,8 +146,16 @@ export const applyRemoteOperations = async < .map((op) => op.id); if (appliedFullStateOpIds.length > 0) { - clearedFullStateOpCount = - await store.clearFullStateOpsExcept(appliedFullStateOpIds); + // Exclude every full-state op of THIS batch, not only the applied ones: + // a later full-state op whose reducers committed but whose archive + // handling failed is still archive_pending/failed and must survive + // cleanup so the retry path can finish it. Deleting it here would let + // markFailed miss it and lose a change already visible at runtime on + // the next startup replay. + const batchFullStateOpIds = appendResult.writtenOps + .filter(isFullStateOperation) + .map((op) => op.id); + clearedFullStateOpCount = await store.clearFullStateOpsExcept(batchFullStateOpIds); } } diff --git a/packages/sync-core/tests/remote-apply.spec.ts b/packages/sync-core/tests/remote-apply.spec.ts index 9aa8e29a06..3c71023caa 100644 --- a/packages/sync-core/tests/remote-apply.spec.ts +++ b/packages/sync-core/tests/remote-apply.spec.ts @@ -112,6 +112,38 @@ describe('applyRemoteOperations', () => { expect(result.clearedFullStateOpCount).toBe(3); }); + it('cleanup preserves a batch full-state op whose archive handling failed (archive_pending)', async () => { + // Both imports reducer-committed; the LATER one failed only its archive + // side effects. Cleanup keyed on the applied one must not delete the + // archive_pending entry, or markFailed misses it and the change is lost + // from the next startup replay. + const appliedImport = createOperation('sync-import-1', 'SYNC_IMPORT'); + const failedImport = createOperation('sync-import-2', 'SYNC_IMPORT'); + const error = new Error('archive failure'); + const store = createStore({ + seqs: [1, 2], + writtenOps: [appliedImport, failedImport], + skippedCount: 0, + }); + const applier = createApplier({ + appliedOps: [appliedImport], + failedOp: { op: failedImport, error }, + }); + + await applyRemoteOperations({ + ops: [appliedImport, failedImport], + store, + applier, + isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT', + }); + + expect(store.clearFullStateOpsExcept).toHaveBeenCalledWith([ + 'sync-import-1', + 'sync-import-2', + ]); + expect(store.markFailed).toHaveBeenCalledWith(['sync-import-2']); + }); + it('checkpoints the whole reducer batch but charges only the attempted archive failure', async () => { const op1 = createOperation('op-1'); const op2 = createOperation('op-2'); diff --git a/src/app/op-log/apply/failed-op-ids.util.spec.ts b/src/app/op-log/apply/failed-op-ids.util.spec.ts deleted file mode 100644 index b3e10d6b09..0000000000 --- a/src/app/op-log/apply/failed-op-ids.util.spec.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { getFailedOpIdsFromBatch } from './failed-op-ids.util'; -import { Operation } from '../core/operation.types'; - -const op = (id: string): Operation => ({ id }) as Operation; - -describe('getFailedOpIdsFromBatch', () => { - it('returns the failed op and every op after it (slice-from-failure)', () => { - const ops = [op('a'), op('b'), op('c'), op('d')]; - expect(getFailedOpIdsFromBatch(ops, op('b'))).toEqual(['b', 'c', 'd']); - }); - - it('returns just the last op when it is the one that failed', () => { - const ops = [op('a'), op('b'), op('c')]; - expect(getFailedOpIdsFromBatch(ops, op('c'))).toEqual(['c']); - }); - - it('returns all ops when the first one failed', () => { - const ops = [op('a'), op('b'), op('c')]; - expect(getFailedOpIdsFromBatch(ops, op('a'))).toEqual(['a', 'b', 'c']); - }); - - it('falls back to only the failed op when it is not in the batch (defensive -1 guard)', () => { - // Guards against slice(-1) wrongly selecting just the last op. - const ops = [op('a'), op('b'), op('c')]; - expect(getFailedOpIdsFromBatch(ops, op('x'))).toEqual(['x']); - }); -}); diff --git a/src/app/op-log/apply/failed-op-ids.util.ts b/src/app/op-log/apply/failed-op-ids.util.ts deleted file mode 100644 index bc74863c8c..0000000000 --- a/src/app/op-log/apply/failed-op-ids.util.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { Operation } from '../core/operation.types'; - -/** - * Given the ops handed to a single `applyOperations()` batch in causal (seq) - * order and the op whose archive side effect threw, return the ids of every op - * that stays unapplied: the failed op plus everything after it. - * - * The batch applier (`replayOperationBatch`) stops at the first failing archive - * side effect, so ops after `failedOp` were never processed and must be retried - * together with it. This is the slice-from-failure handling shared by every - * caller that applies a batch and then marks the remainder failed (the remote - * apply path, conflict resolution, and the failed-op retry on hydration). - * - * The `-1` guard is defensive: `failedOp` always originates from `opsToApply`, - * but if it somehow isn't found we mark only the failed op rather than letting - * `slice(-1)` wrongly select just the last op. - */ -export const getFailedOpIdsFromBatch = ( - opsToApply: Operation[], - failedOp: Operation, -): string[] => { - const failedIndex = opsToApply.findIndex((op) => op.id === failedOp.id); - const stillFailed = failedIndex === -1 ? [failedOp] : opsToApply.slice(failedIndex); - return stillFailed.map((op) => op.id); -}; diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts index 44a57dc990..d395ec010e 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts @@ -8,7 +8,6 @@ import { getDeferredActions, clearDeferredActions, DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD, - DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP, } from './operation-capture.meta-reducer'; import { OperationCaptureService } from './operation-capture.service'; import { Action } from '@ngrx/store'; @@ -329,25 +328,43 @@ describe('operationCaptureMetaReducer', () => { expect(getDeferredActions()).toEqual(actions); }); - it('should retain every accepted action past the pathological warning cap', () => { + it('should fire the reload warning only once per stuck window (no per-action spam)', () => { const confirmSpy = spyNativeDialogs(); - const actions = createManyActions(DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP); - const dropErrorCalls = (): unknown[][] => + const actions = createManyActions( + DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD + 200, + ); + const reloadWarningCalls = (): unknown[][] => confirmSpy.calls .allArgs() - .filter((args) => /pathological cap/.test(String(args[0]))); + .filter((args) => /consider reloading/.test(String(args[0]))); actions.forEach((a) => bufferDeferredAction(a)); - expect(dropErrorCalls().length).toBe(0); - const extraAction = createMockAction({ type: '[Test] Extra Action' }); - bufferDeferredAction(extraAction); - expect(dropErrorCalls().length).toBe(1); + // In dev builds devError opens a blocking dialog; firing it on every + // buffered action past the threshold would freeze the session exactly + // when sync is stuck. + expect(reloadWarningCalls().length).toBe(1); + expect(getDeferredActions()).toEqual(actions); + }); - const buffered = getDeferredActions(); - expect(buffered.length).toBe(DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP + 1); - expect(buffered[0]).toBe(actions[0]); - expect(buffered[buffered.length - 1]).toBe(extraAction); + it('should warn again after the buffer drained and a new stuck window crosses the threshold', () => { + const confirmSpy = spyNativeDialogs(); + const reloadWarningCalls = (): unknown[][] => + confirmSpy.calls + .allArgs() + .filter((args) => /consider reloading/.test(String(args[0]))); + + createManyActions(DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD).forEach((a) => + bufferDeferredAction(a), + ); + expect(reloadWarningCalls().length).toBe(1); + + clearDeferredActions(); + + createManyActions(DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD).forEach((a) => + bufferDeferredAction(a), + ); + expect(reloadWarningCalls().length).toBe(2); }); }); }); diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.ts b/src/app/op-log/capture/operation-capture.meta-reducer.ts index eefd5476ee..775d053909 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.ts @@ -148,41 +148,38 @@ const DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD = 10; export const DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD = 100; /** - * Pathological high-water mark for the deferred actions buffer. ~100 buffered actions are - * plausibly reachable by a real user during a stuck/very long remote-apply - * window, so we must never drop there (see reload-warning threshold above, - * which tells the user to reload long before this mark can be hit through - * real interaction). Crossing it logs loudly but still cannot discard an - * action whose reducer change has already been accepted. - * Exported for tests. + * Highest threshold already warned about in the current stuck window; each + * warning fires once per crossing instead of on every buffered action (in dev + * builds devError opens a blocking dialog, so per-action firing would freeze + * the session exactly when sync is stuck). Reset when the buffer drains. */ -export const DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP = 5000; +let highestWarnedThreshold = 0; /** * Buffers an action for processing after sync completes. * Called by the meta-reducer when a persistent action arrives during sync. + * + * Never drops an accepted persistent action: its reducer mutation already + * exists in NgRx, so removal here would create permanent unsyncable state. */ export const bufferDeferredAction = (action: PersistentAction): void => { - // Never drop an accepted persistent action: its reducer mutation already - // exists in NgRx, so removal here would create permanent unsyncable state. - // The cap remains a loud runaway-loop diagnostic, not a lossy flow-control - // mechanism. - if (deferredActions.length >= DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP) { - devError( - `[operationCaptureMetaReducer] Deferred actions buffer exceeded pathological cap of ${DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP} items. ` + - 'Retaining all actions to preserve sync correctness. Likely a runaway dispatch loop; reload the app.', - ); - } - deferredActions.push(action); deferredActionSet.add(action); - if (deferredActions.length >= DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD) { + if ( + deferredActions.length >= DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD && + highestWarnedThreshold < DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD + ) { + highestWarnedThreshold = DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD; devError( `[operationCaptureMetaReducer] Deferred actions buffer has ${deferredActions.length} items. ` + `Sync may be stuck - consider reloading the app. Nothing is dropped; actions remain buffered.`, ); - } else if (deferredActions.length > DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD) { + } else if ( + deferredActions.length > DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD && + highestWarnedThreshold < DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD + ) { + highestWarnedThreshold = DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD; devError( `[operationCaptureMetaReducer] Deferred actions buffer has ${deferredActions.length} items - sync may be stuck or taking too long`, ); @@ -204,6 +201,9 @@ export const acknowledgeDeferredAction = (action: PersistentAction): void => { } deferredActions.splice(index, 1); deferredActionSet.delete(action); + if (deferredActions.length === 0) { + highestWarnedThreshold = 0; + } }; /** @@ -217,6 +217,7 @@ export const clearDeferredActions = (): void => { deferredActionSet.delete(action); } deferredActions = []; + highestWarnedThreshold = 0; }; /** diff --git a/src/app/op-log/capture/operation-log.effects.spec.ts b/src/app/op-log/capture/operation-log.effects.spec.ts index b8881bb659..cba00d4ac5 100644 --- a/src/app/op-log/capture/operation-log.effects.spec.ts +++ b/src/app/op-log/capture/operation-log.effects.spec.ts @@ -881,29 +881,105 @@ describe('OperationLogEffects', () => { expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); }); - it('should keep an exhausted deferred write queued while acknowledging later successes', async () => { + it('should stop the drain at an exhausted transient failure, keeping it AND its successors queued in order', async () => { + // Persisting a successor before the failed action would record them in + // reversed order with inverted vector clocks — the OLDER same-entity + // edit would win LWW on every client. const failedAction = createPersistentAction(ActionType.TASK_SHARED_ADD); - const successfulAction = createPersistentAction(ActionType.TASK_SHARED_UPDATE); + const successorAction = createPersistentAction(ActionType.TASK_SHARED_UPDATE); bufferDeferredAction(failedAction); - bufferDeferredAction(successfulAction); - let attempts = 0; - mockOpLogStore.appendWithVectorClockUpdate.and.callFake(() => { - attempts++; - return attempts <= 3 - ? Promise.reject(new Error('persistent failure')) - : Promise.resolve(1); - }); + bufferDeferredAction(successorAction); + mockOpLogStore.appendWithVectorClockUpdate.and.rejectWith( + new Error('transient failure'), + ); await effects.processDeferredActions(); - expect(getDeferredActions()).toEqual([failedAction]); + // Only the failed action was attempted (3 retries); the successor was + // never written out of order and both remain buffered. + expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(3); + expect(getDeferredActions()).toEqual([failedAction, successorAction]); + mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(2); await effects.processDeferredActions(); + // Next window drains both in the original order. + const calls = mockOpLogStore.appendWithVectorClockUpdate.calls.all(); + expect(calls.length).toBe(2); + expect(calls[0].args[0].actionType).toBe(ActionType.TASK_SHARED_ADD); + expect(calls[1].args[0].actionType).toBe(ActionType.TASK_SHARED_UPDATE); expect(getDeferredActions()).toEqual([]); }); + it('should abandon a permanently invalid deferred action and continue with its successors', async () => { + // Invalid entity identifiers are deterministic: retrying every sync + // window forever (with a sticky error snack each time) can never succeed. + const invalidAction = createPersistentAction(ActionType.TASK_SHARED_ADD); + (invalidAction.meta as { entityId?: string }).entityId = ''; + const validAction = createPersistentAction(ActionType.TASK_SHARED_UPDATE); + bufferDeferredAction(invalidAction); + bufferDeferredAction(validAction); + + await effects.processDeferredActions(); + + // Invalid action: single attempt, no retries, abandoned. Valid successor + // persisted (its relative order w.r.t. an unpersistable action is moot). + expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect( + mockOpLogStore.appendWithVectorClockUpdate.calls.mostRecent().args[0].actionType, + ).toBe(ActionType.TASK_SHARED_UPDATE); + expect(getDeferredActions()).toEqual([]); + expect(mockSnackService.open).toHaveBeenCalledWith( + jasmine.objectContaining({ msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED }), + ); + }); + + it('should serialize overlapping drains so one buffered action is persisted exactly once', async () => { + // getDeferredActions() is a non-destructive snapshot: without + // serialization two concurrent drains would both see the same + // unacknowledged action and mint two ops for one user intent. + const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); + bufferDeferredAction(action); + + let resolveFirstWrite!: (seq: number) => void; + mockOpLogStore.appendWithVectorClockUpdate.and.returnValue( + new Promise((resolve) => { + resolveFirstWrite = resolve; + }), + ); + + const firstDrain = effects.processDeferredActions(); + const secondDrain = effects.processDeferredActions(); + // Let the first drain reach its (pending) write before releasing it. + await new Promise((resolve) => setTimeout(resolve, 0)); + resolveFirstWrite(1); + await Promise.all([firstDrain, secondDrain]); + + expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(getDeferredActions()).toEqual([]); + }); + + it('should not re-append a deferred action when post-append bookkeeping fails', async () => { + // After appendWithVectorClockUpdate commits, a bookkeeping throw (e.g. + // getCompactionCounter) must not bubble into the retry loop — that would + // append the same user action again under a fresh op id and double-apply + // additive payloads on every client. + const action = createPersistentAction(ActionType.TASK_SHARED_UPDATE); + bufferDeferredAction(action); + mockOpLogStore.getCompactionCounter.and.rejectWith( + new Error('bookkeeping failure'), + ); + + await effects.processDeferredActions(); + + expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalledTimes(1); + expect(getDeferredActions()).toEqual([]); + expect(mockSnackService.open).not.toHaveBeenCalledWith( + jasmine.objectContaining({ msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED }), + ); + }); + it('should use fresh vector clock for deferred actions', async () => { // Set up vector clock to return a specific value mockVectorClockService.getCurrentVectorClock.and.returnValue( diff --git a/src/app/op-log/capture/operation-log.effects.ts b/src/app/op-log/capture/operation-log.effects.ts index 8b68de3511..68c4b70165 100644 --- a/src/app/op-log/capture/operation-log.effects.ts +++ b/src/app/op-log/capture/operation-log.effects.ts @@ -4,7 +4,10 @@ import type { DeferredLocalActionsPort } from '@sp/sync-core'; import { ALL_ACTIONS } from '../../util/local-actions.token'; import { concatMap, filter } from 'rxjs/operators'; import { LockService } from '../sync/lock.service'; -import { LockAcquisitionTimeoutError } from '../core/errors/sync-errors'; +import { + LockAcquisitionTimeoutError, + PermanentDeferredWriteError, +} from '../core/errors/sync-errors'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { isPersistentAction, @@ -63,6 +66,12 @@ export class OperationLogEffects implements DeferredLocalActionsPort { * Initialized lazily from persisted value on first operation. */ private inMemoryCompactionCounter: number | null = null; + + /** + * Serialization chain for processDeferredActions — overlapping drains would + * double-persist the same buffered action (see processDeferredActions). + */ + private _deferredProcessingChain: Promise = Promise.resolve(); /** * Dedupe timestamp for the storage-quota snackbar. #7700: when quota fires * inside the deferred-action retry loop, the retry loop calls handleQuotaExceeded @@ -182,12 +191,17 @@ export class OperationLogEffects implements DeferredLocalActionsPort { if (!isBulkAllOperation && !hasValidEntityId && !hasValidEntityIds) { // No queue bookkeeping needed here: the effect wrapper's `finally` // decrements the pending counter for this action even on early return. + if (isDeferredWrite) { + // Typed throw BEFORE devError: in dev builds devError itself can throw + // a generic Error, which the deferred drain would misclassify as a + // transient (retryable) failure. The drain logs the abandonment loudly. + throw new PermanentDeferredWriteError( + `Deferred action ${action.type} has invalid entity identifiers.`, + ); + } devError( `[OperationLogEffects] Action ${action.type} has invalid entityId/entityIds (${action.meta.entityId}) - skipping persistence`, ); - if (isDeferredWrite) { - throw new Error(`Deferred action ${action.type} has invalid entity identifiers.`); - } return; } @@ -300,7 +314,9 @@ export class OperationLogEffects implements DeferredLocalActionsPort { }, }); if (isDeferredWrite) { - throw new Error(`Deferred action ${action.type} has an invalid payload.`); + throw new PermanentDeferredWriteError( + `Deferred action ${action.type} has an invalid payload.`, + ); } return; // Skip persisting invalid operation } @@ -320,32 +336,44 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // The op.vectorClock already contains the incremented clock (from newClock above). await this.opLogStore.appendWithVectorClockUpdate(op, 'local'); - // Mark that we have pending ops (not yet uploaded) for UI indicator - this.superSyncStatusService.updatePendingOpsStatus(true); + // The op is durably committed past this point. Bookkeeping failures + // below must NOT propagate: a throw would send the deferred retry loop + // back through writeOperation, appending the same user action again + // under a fresh op id — additive payloads (time tracking, counters) + // would then double-apply on every client. + try { + // Mark that we have pending ops (not yet uploaded) for UI indicator + this.superSyncStatusService.updatePendingOpsStatus(true); - // Track write count for high-volume debugging - this.writeCount++; - if (this.writeCount % 50 === 0) { - OpLog.normal( - `OperationLogEffects: Wrote ${this.writeCount} operations to IndexedDB`, + // Track write count for high-volume debugging + this.writeCount++; + if (this.writeCount % 50 === 0) { + OpLog.normal( + `OperationLogEffects: Wrote ${this.writeCount} operations to IndexedDB`, + ); + } + + // 1b. Trigger immediate upload to SuperSync (async, non-blocking) + this.immediateUploadService.trigger(); + + // 2. Check if compaction is needed + // PERF: Use in-memory counter instead of IndexedDB transaction on every operation. + // Initialize from persisted value on first use (for crash recovery). + if (this.inMemoryCompactionCounter === null) { + this.inMemoryCompactionCounter = await this.opLogStore.getCompactionCounter(); + } + this.inMemoryCompactionCounter++; + if (this.inMemoryCompactionCounter >= COMPACTION_THRESHOLD) { + // Trigger compaction asynchronously (don't block write operation) + // Counter is reset in compaction service on success + this.triggerCompaction(); + } + } catch (bookkeepingError) { + OpLog.err( + 'OperationLogEffects: Post-append bookkeeping failed (operation is already durable; not retrying the append)', + { name: (bookkeepingError as Error | undefined)?.name }, ); } - - // 1b. Trigger immediate upload to SuperSync (async, non-blocking) - this.immediateUploadService.trigger(); - - // 2. Check if compaction is needed - // PERF: Use in-memory counter instead of IndexedDB transaction on every operation. - // Initialize from persisted value on first use (for crash recovery). - if (this.inMemoryCompactionCounter === null) { - this.inMemoryCompactionCounter = await this.opLogStore.getCompactionCounter(); - } - this.inMemoryCompactionCounter++; - if (this.inMemoryCompactionCounter >= COMPACTION_THRESHOLD) { - // Trigger compaction asynchronously (don't block write operation) - // Counter is reset in compaction service on success - this.triggerCompaction(); - } }; if (options.callerHoldsOperationLogLock) { @@ -365,6 +393,11 @@ export class OperationLogEffects implements DeferredLocalActionsPort { } catch (e) { // 4.1.1 Error Handling for Optimistic Updates OpLog.err('OperationLogEffects: Failed to persist operation', e); + if (e instanceof PermanentDeferredWriteError) { + // Deterministic invalidity, already surfaced via its own snack — + // rethrow untouched so the deferred drain abandons instead of retrying. + throw e; + } if (e instanceof LockAcquisitionTimeoutError) { // #7700: do NOT silently swallow lock timeouts. Pre-fix, a reentrant // sp_op_log timeout was caught here, the user got a snackbar, and the @@ -620,6 +653,28 @@ export class OperationLogEffects implements DeferredLocalActionsPort { * Called after sync operations are applied and remote op bookkeeping is complete. */ async processDeferredActions(options: WriteOperationOptions = {}): Promise { + // Serialize overlapping invocations: getDeferredActions() is a + // non-destructive snapshot, so two concurrent drains (e.g. a WS-triggered + // download finishing while a piggyback apply flushes — they run under + // different locks) would both see the same unacknowledged action and each + // mint an op with a fresh id for it. One user intent = one op; additive + // payloads would otherwise double-apply on every client. Chaining makes + // the second caller wait and re-snapshot after the first finishes. + const run = this._deferredProcessingChain.then(() => + this._processDeferredActionsImpl(options), + ); + // Keep the chain alive even if this run rejects; errors still surface to + // this invocation's caller via `run`. + this._deferredProcessingChain = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + private async _processDeferredActionsImpl( + options: WriteOperationOptions, + ): Promise { const deferredActions = getDeferredActions(); if (deferredActions.length === 0) { return; @@ -647,6 +702,10 @@ export class OperationLogEffects implements DeferredLocalActionsPort { break; } catch (e) { lastError = e; + if (e instanceof PermanentDeferredWriteError) { + // Deterministic invalidity — retrying cannot succeed. + break; + } if (attempt < MAX_RETRIES - 1) { // Exponential backoff: 100ms, 200ms, 400ms const delay = BASE_DELAY_MS * Math.pow(2, attempt); @@ -659,19 +718,38 @@ export class OperationLogEffects implements DeferredLocalActionsPort { } } - if (!success) { - failedCount++; - // Log error after all retries exhausted, continue processing remaining actions - OpLog.err( - `OperationLogEffects: Failed to process deferred action after ${MAX_RETRIES} retries`, - { - actionType: action.type, - errorName: (lastError as Error | undefined)?.name, - }, - ); - } else { + if (success) { acknowledgeDeferredAction(action); + continue; } + + failedCount++; + + if (lastError instanceof PermanentDeferredWriteError) { + // Terminal: the action can never be persisted (invalid identifiers or + // payload). Abandon it — keeping it queued would retry it with backoff + // and re-raise the failure snack on every future sync window while the + // in-memory buffer's "durability" ends at the advised reload anyway. + acknowledgeDeferredAction(action); + OpLog.err('OperationLogEffects: Abandoning permanently invalid deferred action', { + actionType: action.type, + }); + continue; + } + + // Transient failure: STOP the drain here instead of persisting later + // actions first. Writing a successor now and this action on a future + // drain would record them in reversed order with inverted vector clocks, + // so the OLDER same-entity edit would win LWW on every client. The whole + // suffix stays buffered and is retried in order on the next window. + OpLog.err( + `OperationLogEffects: Deferred action failed after ${MAX_RETRIES} retries; keeping it and all later actions queued in order`, + { + actionType: action.type, + errorName: (lastError as Error | undefined)?.name, + }, + ); + break; } // Show notification if any actions failed diff --git a/src/app/op-log/core/errors/sync-errors.ts b/src/app/op-log/core/errors/sync-errors.ts index b7fa8d71a8..f6f33e9002 100644 --- a/src/app/op-log/core/errors/sync-errors.ts +++ b/src/app/op-log/core/errors/sync-errors.ts @@ -112,6 +112,16 @@ export class UnknownSyncStateError extends Error { override name = 'UnknownSyncStateError'; } +/** + * A deferred action can never be persisted (invalid entity identifiers or an + * invalid operation payload) — a deterministic condition, not a transient + * I/O failure. processDeferredActions abandons such actions instead of + * retrying them on every sync window forever. + */ +export class PermanentDeferredWriteError extends Error { + override name = 'PermanentDeferredWriteError'; +} + // -----ENCRYPTION & COMPRESSION---- export class DecryptNoPasswordError extends AdditionalLogErrorBase { override name = 'DecryptNoPasswordError'; diff --git a/src/app/op-log/persistence/db-keys.const.ts b/src/app/op-log/persistence/db-keys.const.ts index 6d96c32a41..7d59f6319f 100644 --- a/src/app/op-log/persistence/db-keys.const.ts +++ b/src/app/op-log/persistence/db-keys.const.ts @@ -45,6 +45,15 @@ export const BACKUP_KEY = 'backup' as const; /** Meta key for derived full-state operation refs */ export const FULL_STATE_OPS_META_KEY = 'full_state_ops' as const; +/** + * Meta key marking an interrupted USE_REMOTE raw rebuild: set atomically with + * the baseline replacement, cleared after the server-history replay commits. + * While set, the next sync must redo the raw (own-ops-included) rebuild — + * the normal download path excludes this client's own ops server-side, so an + * interrupted replay would otherwise silently lose them. + */ +export const RAW_REBUILD_INCOMPLETE_META_KEY = 'raw_rebuild_incomplete' as const; + /** Index names for ops object store */ export const OPS_INDEXES = { BY_ID: 'byId' as const, diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts index 308be9126e..815ae29a50 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts @@ -410,6 +410,37 @@ describe('OperationLogHydratorService', () => { expect(mockHydrationStateService.endApplyingRemoteOps).toHaveBeenCalled(); }); + it('should replay a tail op with a malformed stored schemaVersion verbatim instead of failing into recovery', async () => { + // Strict schemaVersion parsing guards the receive/upload paths; on the + // local hydration path one legacy/corrupt entry must not turn every + // boot into attemptRecovery() (which can lose tail data). + const snapshot = createMockSnapshot({ lastAppliedOpSeq: 5 }); + const malformedOp = createMockOperation('op-6', OpType.Update, { + schemaVersion: '2' as unknown as number, + }); + const tailOps = [ + createMockEntry(6, malformedOp), + createMockEntry(7, createMockOperation('op-7')), + ]; + mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); + mockOpLogStore.getOpsAfterSeq.and.returnValue(Promise.resolve(tailOps)); + + await service.hydrateStore(); + + expect(mockRecoveryService.attemptRecovery).not.toHaveBeenCalled(); + // Both ops replay in order; the malformed one is passed through with a + // sanitized version stamp (payload untouched). + expect(mockStore.dispatch).toHaveBeenCalledWith( + bulkApplyHydrationOperations({ + operations: [ + { ...malformedOp, schemaVersion: CURRENT_SCHEMA_VERSION }, + tailOps[1].op, + ], + localClientId: 'test-client', + }), + ); + }); + it('should apply a failed tail op exactly once across hydration replay + retry (one boot)', async () => { // Regression: a remote op marked 'failed' (archive side effect threw // after its reducer committed) with seq > lastAppliedOpSeq used to get @@ -993,10 +1024,12 @@ describe('OperationLogHydratorService', () => { lastAppliedOpSeq: 5, }); + // schemaVersion 1 = a legitimately old version (the floor); 0 would be + // malformed and is sanitized by the lenient hydration boundary instead. const tailOps = [ createMockEntry( 6, - createMockOperation('op-6', OpType.Update, { schemaVersion: 0 }), // Old + createMockOperation('op-6', OpType.Update, { schemaVersion: 1 }), // Old ), createMockEntry( 7, @@ -1006,7 +1039,7 @@ describe('OperationLogHydratorService', () => { ), createMockEntry( 8, - createMockOperation('op-8', OpType.Update, { schemaVersion: 0 }), // Old + createMockOperation('op-8', OpType.Update, { schemaVersion: 1 }), // Old ), ]; @@ -1015,7 +1048,7 @@ describe('OperationLogHydratorService', () => { mockSchemaMigrationService.needsMigration.and.returnValue(false); // At least one operation needs migration mockSchemaMigrationService.operationNeedsMigration.and.callFake( - (op: Operation) => op.schemaVersion === 0, + (op: Operation) => op.schemaVersion === 1, ); await service.hydrateStore(); diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.ts b/src/app/op-log/persistence/operation-log-hydrator.service.ts index e972905af4..05b86adfe2 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.ts @@ -5,6 +5,7 @@ import { loadAllData } from '../../root-store/meta/load-all-data.action'; import { OperationLogMigrationService } from './operation-log-migration.service'; import { CURRENT_SCHEMA_VERSION, + getOperationSchemaVersion, SchemaMigrationService, } from './schema-migration.service'; import { OperationLogSnapshotService } from './operation-log-snapshot.service'; @@ -484,20 +485,39 @@ export class OperationLogHydratorService { * @returns Array of migrated operations */ private _migrateTailOps(ops: Operation[]): Operation[] { + // Lenient boundary: a malformed stored schemaVersion (legacy or corrupt + // entry) must not abort the WHOLE hydration into attemptRecovery() — that + // trades one questionable op for possible tail-data loss on every boot. + // Strict parsing stays on the receive/upload paths; locally we replay the + // op verbatim as a best effort (stamping the current version so + // migrateOperations passes it through unchanged, preserving order). + const sanitizedOps = ops.map((op) => { + try { + getOperationSchemaVersion(op); + return op; + } catch { + OpLog.warn( + 'OperationLogHydratorService: Stored op has a malformed schemaVersion; replaying verbatim without migration.', + { id: op.id }, + ); + return { ...op, schemaVersion: CURRENT_SCHEMA_VERSION }; + } + }); + // Check if any ops need migration - const needsMigration = ops.some((op) => + const needsMigration = sanitizedOps.some((op) => this.schemaMigrationService.operationNeedsMigration(op), ); if (!needsMigration) { - return ops; + return sanitizedOps; } OpLog.normal( - `OperationLogHydratorService: Migrating ${ops.length} tail ops to current schema version...`, + `OperationLogHydratorService: Migrating ${sanitizedOps.length} tail ops to current schema version...`, ); - return this.schemaMigrationService.migrateOperations(ops); + return this.schemaMigrationService.migrateOperations(sanitizedOps); } /** @@ -619,9 +639,12 @@ export class OperationLogHydratorService { .filter((seq): seq is number => seq !== undefined); if (appliedSeqs.length > 0) { await this.opLogStore.markApplied(appliedSeqs); - // Parity with the primary remote-apply path: applyRemoteOperations only - // merges clocks for ops it marked applied, so a failed op's clock is - // merged here, when its application completes. + // The primary remote-apply path (applyRemoteOperations) merges clocks at + // reducer commit for the WHOLE batch, including ops whose archive + // handling later fails — so these clocks were usually merged already. + // Re-merging here is a harmless component-wise max and also covers ops + // that reached `failed`/`archive_pending` via crash recovery, where the + // reducer-commit callback (and its clock merge) may never have run. await this.opLogStore.mergeRemoteOpClocks(result.appliedOps); OpLog.normal( `OperationLogHydratorService: Successfully retried ${appliedSeqs.length} failed ops`, 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 2f1da31953..422d7e8a97 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 @@ -2384,6 +2384,26 @@ describe('OperationLogStoreService', () => { ); }); + it('sets the raw-rebuild-incomplete marker atomically with the replacement and clears it on demand', async () => { + expect(await service.isRawRebuildIncomplete()).toBe(false); + + await service.runRemoteStateReplacement({ + baselineState: { task: { ids: [], entities: {} } }, + vectorClock: { remote: 1 }, + schemaVersion: 4, + snapshotEntityKeys: [], + archiveYoung: createArchive('young'), + archiveOld: createArchive('old'), + }); + + // A crash after the replacement but before the replay commits must leave + // the marker set so the next sync redoes the raw rebuild. + expect(await service.isRawRebuildIncomplete()).toBe(true); + + await service.clearRawRebuildIncomplete(); + expect(await service.isRawRebuildIncomplete()).toBe(false); + }); + it('rolls back every store if one archive write fails', async () => { const priorOp = createTestOperation({ entityId: 'prior-task' }); const priorState = { sentinel: 'prior-state' }; diff --git a/src/app/op-log/persistence/operation-log-store.service.ts b/src/app/op-log/persistence/operation-log-store.service.ts index 0db53f6a65..1b387219ba 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -18,6 +18,7 @@ import { SINGLETON_KEY, BACKUP_KEY, FULL_STATE_OPS_META_KEY, + RAW_REBUILD_INCOMPLETE_META_KEY, OPS_INDEXES, ArchiveStoreEntry, ProfileDataStoreEntry, @@ -1577,6 +1578,15 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + await this._ensureInit(); + const entry = await this._adapter.get<{ incomplete?: boolean }>( + STORE_NAMES.META, + RAW_REBUILD_INCOMPLETE_META_KEY, + ); + return entry?.incomplete === true; + } + + /** Marks the USE_REMOTE raw rebuild as fully completed. */ + async clearRawRebuildIncomplete(): Promise { + await this._ensureInit(); + await this._adapter.delete(STORE_NAMES.META, RAW_REBUILD_INCOMPLETE_META_KEY); + } + // ============================================================ // Vector Clock Management (Performance Optimization) // ============================================================ diff --git a/src/app/op-log/persistence/schema-migration.service.ts b/src/app/op-log/persistence/schema-migration.service.ts index 4fbaaafc5a..a154f2f7ad 100644 --- a/src/app/op-log/persistence/schema-migration.service.ts +++ b/src/app/op-log/persistence/schema-migration.service.ts @@ -3,7 +3,6 @@ import { Operation, VectorClock } from '../core/operation.types'; import { OpLog } from '../../core/log'; import { CURRENT_SCHEMA_VERSION as SHARED_CURRENT_SCHEMA_VERSION, - MAX_VERSION_SKIP as SHARED_MAX_VERSION_SKIP, MIN_SUPPORTED_SCHEMA_VERSION as SHARED_MIN_SUPPORTED_SCHEMA_VERSION, MIGRATIONS, migrateState, @@ -17,7 +16,6 @@ import { // Re-export shared constants for backwards compatibility export const CURRENT_SCHEMA_VERSION = SHARED_CURRENT_SCHEMA_VERSION; -export const MAX_VERSION_SKIP = SHARED_MAX_VERSION_SKIP; export const MIN_SUPPORTED_SCHEMA_VERSION = SHARED_MIN_SUPPORTED_SCHEMA_VERSION; // Re-export types @@ -30,11 +28,11 @@ export const getOperationSchemaVersion = (op: { schemaVersion?: unknown }): numb if ( typeof op.schemaVersion !== 'number' || !Number.isInteger(op.schemaVersion) || - op.schemaVersion < 0 + // Floor of 1 matches the server contract (SuperSyncOperationSchema min(1)) + // so the layers cannot drift apart. + op.schemaVersion < 1 ) { - throw new Error( - 'Operation schemaVersion must be a non-negative integer when present.', - ); + throw new Error('Operation schemaVersion must be a positive integer when present.'); } return op.schemaVersion; }; 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 30ea0d7307..7f38e0de66 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 @@ -70,6 +70,9 @@ describe('OperationLogSyncService', () => { 'appendBatchSkipDuplicates', 'hasSyncedOps', 'runRemoteStateReplacement', + 'isRawRebuildIncomplete', + 'clearRawRebuildIncomplete', + 'loadImportBackup', ]); opLogStoreSpy.hasSyncedOps.and.resolveTo(true); opLogStoreSpy.markSynced.and.resolveTo(); @@ -82,6 +85,9 @@ describe('OperationLogSyncService', () => { skippedCount: 0, }); opLogStoreSpy.runRemoteStateReplacement.and.resolveTo(); + opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(false); + opLogStoreSpy.clearRawRebuildIncomplete.and.resolveTo(); + opLogStoreSpy.loadImportBackup.and.resolveTo(null); schemaMigrationServiceSpy = jasmine.createSpyObj('SchemaMigrationService', [ 'getCurrentVersion', @@ -147,8 +153,16 @@ describe('OperationLogSyncService', () => { writeFlushServiceSpy = jasmine.createSpyObj('OperationWriteFlushService', [ 'flushPendingWrites', + 'flushThenRunExclusive', ]); writeFlushServiceSpy.flushPendingWrites.and.resolveTo(); + // Mirror the real barrier semantics: flush BEFORE the exclusive section runs. + writeFlushServiceSpy.flushThenRunExclusive.and.callFake( + async (fn: () => Promise) => { + await writeFlushServiceSpy.flushPendingWrites(); + return fn(); + }, + ); superSyncStatusServiceSpy = jasmine.createSpyObj('SuperSyncStatusService', [ 'updatePendingOpsStatus', @@ -879,6 +893,27 @@ describe('OperationLogSyncService', () => { }); describe('downloadRemoteOps', () => { + it('should redo the raw rebuild when a prior USE_REMOTE replay was interrupted', async () => { + // The normal download path excludes this client's own ops server-side, + // so resuming an interrupted rebuild through it would silently lose them. + opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(true); + const forceDownloadSpy = spyOn( + service, + 'forceDownloadRemoteState', + ).and.resolveTo(); + const mockProvider = { + isReady: () => Promise.resolve(true), + } as any; + + const result = await service.downloadRemoteOps(mockProvider); + + expect(forceDownloadSpy).toHaveBeenCalledWith(mockProvider, { + isCrashResume: true, + }); + expect(result.kind).toBe('snapshot_hydrated'); + expect(downloadServiceSpy.downloadRemoteOps).not.toHaveBeenCalled(); + }); + it('should return localWinOpsCreated: 0 and newOpsCount: 0 when no new ops', async () => { downloadServiceSpy.downloadRemoteOps.and.returnValue( Promise.resolve({ @@ -2533,6 +2568,93 @@ describe('OperationLogSyncService', () => { expect(backupServiceSpy.captureImportBackup).toHaveBeenCalled(); }); + it('should clear the raw-rebuild-incomplete marker only after the replay committed', async () => { + const callOrder: string[] = []; + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); + remoteOpsProcessingServiceSpy.processRemoteOps.and.callFake(async () => { + callOrder.push('processRemoteOps'); + return { + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, + }; + }); + opLogStoreSpy.clearRawRebuildIncomplete.and.callFake(async () => { + callOrder.push('clearRawRebuildIncomplete'); + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as any; + + await service.forceDownloadRemoteState(mockProvider); + + expect(callOrder).toEqual(['processRemoteOps', 'clearRawRebuildIncomplete']); + }); + + it('should NOT clear the raw-rebuild-incomplete marker when the replay is blocked', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); + remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({ + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as any; + + await expectAsync(service.forceDownloadRemoteState(mockProvider)).toBeRejected(); + + expect(opLogStoreSpy.clearRawRebuildIncomplete).not.toHaveBeenCalled(); + }); + + it('should keep the first attempt backup on crash resume instead of re-capturing', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); + opLogStoreSpy.loadImportBackup.and.resolveTo({ state: {}, savedAt: 12345 }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as any; + + await service.forceDownloadRemoteState(mockProvider, { isCrashResume: true }); + + // Re-capturing would overwrite the single backup slot with the partial + // baseline; the original pre-replace snapshot must survive the resume. + expect(backupServiceSpy.captureImportBackup).not.toHaveBeenCalled(); + expect(snackServiceSpy.open).toHaveBeenCalledWith( + jasmine.objectContaining({ + msg: T.F.SYNC.S.LOCAL_DATA_REPLACE_UNDO, + actionStr: T.G.UNDO, + }), + ); + }); + it('should offer to restore the previous data after replacing (#8107)', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [makeRemoteOp()], diff --git a/src/app/op-log/sync/operation-log-sync.service.ts b/src/app/op-log/sync/operation-log-sync.service.ts index 6d97e0d7ce..3050d00754 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -51,8 +51,6 @@ import { SyncLocalStateService } from './sync-local-state.service'; import { SyncImportConflictCoordinatorService } from './sync-import-conflict-coordinator.service'; import { isExampleTaskCreateOp } from '../validation/is-example-task-op.util'; import { Operation } from '../core/operation.types'; -import { LockService } from './lock.service'; -import { LOCK_NAMES } from '../core/operation-log.const'; import { ValidateStateService } from '../validation/validate-state.service'; import { extractEntityKeysFromState } from '../persistence/extract-entity-keys'; @@ -132,7 +130,6 @@ export class OperationLogSyncService { private superSyncStatusService = inject(SuperSyncStatusService); private serverMigrationService = inject(ServerMigrationService); private writeFlushService = inject(OperationWriteFlushService); - private lockService = inject(LockService); private schemaMigrationService = inject(SchemaMigrationService); private validateStateService = inject(ValidateStateService); @@ -439,6 +436,19 @@ export class OperationLogSyncService { syncProvider: OperationSyncCapable, options?: { forceFromSeq0?: boolean; isNeverSynced?: boolean }, ): Promise { + // Crash-resume: a prior USE_REMOTE rebuild committed its baseline + // replacement but crashed before the replay finished. The normal download + // path excludes this client's own ops server-side, so resuming through it + // would silently lose them — redo the raw rebuild instead. + if (await this.opLogStore.isRawRebuildIncomplete()) { + OpLog.warn( + 'OperationLogSyncService: Interrupted USE_REMOTE rebuild detected — redoing the raw rebuild.', + ); + await this.forceDownloadRemoteState(syncProvider, { isCrashResume: true }); + // State was replaced wholesale, exactly like a snapshot hydration. + return { kind: 'snapshot_hydrated' }; + } + const result = await this.downloadService.downloadRemoteOps(syncProvider, options); // FIX #6571: Check download success before processing results. @@ -1174,7 +1184,18 @@ export class OperationLogSyncService { * * @param syncProvider - The sync provider to download from */ - async forceDownloadRemoteState(syncProvider: OperationSyncCapable): Promise { + async forceDownloadRemoteState( + syncProvider: OperationSyncCapable, + options?: { + /** + * Resuming an interrupted rebuild (crash between the baseline replacement + * and the replay commit). Keeps the FIRST attempt's pre-replace safety + * backup: the single backup slot still holds the user's original data, + * and re-capturing here would overwrite it with the partial baseline. + */ + isCrashResume?: boolean; + }, + ): Promise { OpLog.warn( 'OperationLogSyncService: Force downloading remote state for a full rebuild.', ); @@ -1255,15 +1276,20 @@ export class OperationLogSyncService { let capturedBackupSavedAt: number | undefined; let replacementCommitted = false; - let backupSavedAt: number; + let backupSavedAt: number | undefined; try { - backupSavedAt = await this.lockService.request( - LOCK_NAMES.OPERATION_LOG, - async () => { - // Include actions dispatched while the network request and preflight were - // in flight in the reversible safety backup. - await this.writeFlushService.flushPendingWrites(); - let savedAt: number; + // flushThenRunExclusive drains the capture pipeline BEFORE acquiring the + // op-log lock (flushPendingWrites re-acquires the same non-reentrant lock, + // so flushing while holding it deadlocks) and re-checks inside the lock — + // actions dispatched while the network request and preflight were in + // flight are durably written and included in the reversible safety backup. + backupSavedAt = await this.writeFlushService.flushThenRunExclusive(async () => { + let savedAt: number | undefined; + if (options?.isCrashResume) { + // Keep the first attempt's pre-replace backup (see option JSDoc). + savedAt = (await this.opLogStore.loadImportBackup())?.savedAt; + capturedBackupSavedAt = savedAt; + } else { try { savedAt = await this.backupService.captureImportBackup(); capturedBackupSavedAt = savedAt; @@ -1276,121 +1302,129 @@ export class OperationLogSyncService { 'Pre-replace safety backup failed; aborting to preserve local state.', ); } + } - // 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); - await this.opLogStore.runRemoteStateReplacement({ - baselineState, - vectorClock: rebuiltClock, - schemaVersion: CURRENT_SCHEMA_VERSION, - snapshotEntityKeys: extractEntityKeysFromState( - baselineState as Parameters[0], - ), - archiveYoung, - archiveOld, - }); - replacementCommitted = true; + // 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); + await this.opLogStore.runRemoteStateReplacement({ + baselineState, + vectorClock: rebuiltClock, + schemaVersion: CURRENT_SCHEMA_VERSION, + snapshotEntityKeys: extractEntityKeysFromState( + baselineState as Parameters[0], + ), + archiveYoung, + archiveOld, + }); + replacementCommitted = true; + OpLog.normal( + 'OperationLogSyncService: Replaced local persistence with remote baseline.', + ); + + // FILE-BASED SYNC: Handle snapshot state from force download. + // When downloading from seq 0 on file-based providers, we may receive a + // snapshotState instead of incremental ops. This happens when the remote + // has a SYNC_IMPORT (full state snapshot) with empty recentOps. + // hydrateFromRemoteSync persists its own state cache + vector clock. + if (result.providerMode === 'fileSnapshotOps' && snapshotState) { OpLog.normal( - 'OperationLogSyncService: Replaced local persistence with remote baseline.', + 'OperationLogSyncService: Force download received snapshotState. Hydrating...', ); - // FILE-BASED SYNC: Handle snapshot state from force download. - // When downloading from seq 0 on file-based providers, we may receive a - // snapshotState instead of incremental ops. This happens when the remote - // has a SYNC_IMPORT (full state snapshot) with empty recentOps. - // hydrateFromRemoteSync persists its own state cache + vector clock. - if (result.providerMode === 'fileSnapshotOps' && snapshotState) { + // Hydrate from snapshot - DON'T create SYNC_IMPORT since we're + // accepting remote state, not uploading local state. + await this.syncHydrationService.hydrateFromRemoteSync( + snapshotState, + result.snapshotVectorClock, + 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', + ); OpLog.normal( - 'OperationLogSyncService: Force download received snapshotState. Hydrating...', - ); - - // Hydrate from snapshot - DON'T create SYNC_IMPORT since we're - // accepting remote state, not uploading local state. - await this.syncHydrationService.hydrateFromRemoteSync( - snapshotState, - result.snapshotVectorClock, - 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', - ); - OpLog.normal( - `OperationLogSyncService: Wrote ${appendResult.writtenOps.length} snapshot ops to IndexedDB ` + - 'after force-download hydration.' + - (appendResult.skippedCount > 0 - ? ` Skipped ${appendResult.skippedCount} duplicate(s).` - : ''), - ); - } - - // Update lastServerSeq after hydration - if (result.latestServerSeq !== undefined) { - await syncProvider.setLastServerSeq(result.latestServerSeq); - } - - OpLog.normal( - 'OperationLogSyncService: Force download snapshot hydration complete.', - ); - return savedAt; - } - - // Reset live state to defaults, then replay the COMPLETE server history on - // top. A full-state op in the history replaces state again by its own - // semantics; a purely incremental history rebuilds from this baseline. - this.store.dispatch( - loadAllData({ - appDataComplete: defaultData as Parameters< - typeof loadAllData - >[0]['appDataComplete'], - }), - ); - // Brief yield to let NgRx process the state reset - await new Promise((resolve) => setTimeout(resolve, 0)); - - // Process all remote ops (no confirmation needed - user already chose USE_REMOTE). - // Skip conflict detection because the NgRx store was just reset to empty state, - // which causes all entities to appear missing and CONCURRENT ops to be discarded. - // Validation failure is surfaced via the session-validation latch. (#7330) - const processResult = await this.remoteOpsProcessingService.processRemoteOps( - migratedRemoteOps, - { - skipConflictDetection: true, - callerHoldsOperationLogLock: true, - }, - ); - - if (processResult.blockedByIncompatibleOp) { - // Version blocks were pre-checked above; only a migration exception lands - // here. The rebuild is partial: keep the cursor at 0 so the next sync - // retries the remainder, and surface the failure — the Undo snack still - // offers the pre-replace backup. - throw new Error( - 'USE_REMOTE incomplete: an op failed schema migration during replay.', + `OperationLogSyncService: Wrote ${appendResult.writtenOps.length} snapshot ops to IndexedDB ` + + 'after force-download hydration.' + + (appendResult.skippedCount > 0 + ? ` Skipped ${appendResult.skippedCount} duplicate(s).` + : ''), ); } - // Update lastServerSeq + // Update lastServerSeq after hydration if (result.latestServerSeq !== undefined) { await syncProvider.setLastServerSeq(result.latestServerSeq); } + // Replay committed — the rebuild is complete and crash-resume is + // no longer needed. + await this.opLogStore.clearRawRebuildIncomplete(); + OpLog.normal( - `OperationLogSyncService: Force download complete. Rebuilt from ${migratedRemoteOps.length} ops.`, + 'OperationLogSyncService: Force download snapshot hydration complete.', ); return savedAt; - }, - ); + } + + // Reset live state to defaults, then replay the COMPLETE server history on + // top. A full-state op in the history replaces state again by its own + // semantics; a purely incremental history rebuilds from this baseline. + this.store.dispatch( + loadAllData({ + appDataComplete: defaultData as Parameters< + typeof loadAllData + >[0]['appDataComplete'], + }), + ); + // Brief yield to let NgRx process the state reset + await new Promise((resolve) => setTimeout(resolve, 0)); + + // Process all remote ops (no confirmation needed - user already chose USE_REMOTE). + // Skip conflict detection because the NgRx store was just reset to empty state, + // which causes all entities to appear missing and CONCURRENT ops to be discarded. + // Validation failure is surfaced via the session-validation latch. (#7330) + const processResult = await this.remoteOpsProcessingService.processRemoteOps( + migratedRemoteOps, + { + skipConflictDetection: true, + callerHoldsOperationLogLock: true, + }, + ); + + if (processResult.blockedByIncompatibleOp) { + // Version blocks were pre-checked above; only a migration exception lands + // here. The rebuild is partial: keep the cursor at 0 so the next sync + // retries the remainder, and surface the failure — the Undo snack still + // offers the pre-replace backup. + throw new Error( + 'USE_REMOTE incomplete: an op failed schema migration during replay.', + ); + } + + // Update lastServerSeq + if (result.latestServerSeq !== undefined) { + await syncProvider.setLastServerSeq(result.latestServerSeq); + } + + // Replay committed — the rebuild is complete and crash-resume is + // no longer needed. + await this.opLogStore.clearRawRebuildIncomplete(); + + OpLog.normal( + `OperationLogSyncService: Force download complete. Rebuilt from ${migratedRemoteOps.length} ops.`, + ); + return savedAt; + }); } catch (e) { if (replacementCommitted && capturedBackupSavedAt !== undefined) { this._showRestorePreviousDataSnack(capturedBackupSavedAt); @@ -1398,7 +1432,10 @@ export class OperationLogSyncService { throw e; } - this._showRestorePreviousDataSnack(backupSavedAt); + // On a crash resume without a surviving backup there is nothing to offer. + if (backupSavedAt !== undefined) { + this._showRestorePreviousDataSnack(backupSavedAt); + } } private _preflightRemoteOperations(remoteOps: Operation[]): Operation[] { @@ -1406,9 +1443,16 @@ export class OperationLogSyncService { let version: number; try { version = getOperationSchemaVersion(op as { schemaVersion?: unknown }); - } catch { + } catch (e) { + // Keep the root cause diagnosable (id-only, no payloads) — this is a + // rare, support-heavy failure path. + OpLog.err('OperationLogSyncService: USE_REMOTE preflight version parse failed', { + id: op.id, + name: (e as Error | undefined)?.name, + }); throw new Error( 'USE_REMOTE aborted: remote history has an invalid schema version.', + { cause: e }, ); } @@ -1433,8 +1477,13 @@ export class OperationLogSyncService { try { return this.schemaMigrationService.migrateOperations(remoteOps); - } catch { - throw new Error('USE_REMOTE aborted: remote operation migration failed.'); + } catch (e) { + OpLog.err('OperationLogSyncService: USE_REMOTE preflight migration failed', { + name: (e as Error | undefined)?.name, + }); + throw new Error('USE_REMOTE aborted: remote operation migration failed.', { + cause: e, + }); } } diff --git a/src/app/op-log/sync/operation-write-flush.service.spec.ts b/src/app/op-log/sync/operation-write-flush.service.spec.ts index 9052c23735..a868d129a8 100644 --- a/src/app/op-log/sync/operation-write-flush.service.spec.ts +++ b/src/app/op-log/sync/operation-write-flush.service.spec.ts @@ -1,21 +1,28 @@ import { TestBed } from '@angular/core/testing'; import { OperationWriteFlushService } from './operation-write-flush.service'; import { LockService } from './lock.service'; +import { OperationCaptureService } from '../capture/operation-capture.service'; describe('OperationWriteFlushService', () => { let service: OperationWriteFlushService; let lockServiceSpy: jasmine.SpyObj; + let captureServiceSpy: jasmine.SpyObj; beforeEach(() => { lockServiceSpy = jasmine.createSpyObj('LockService', ['request']); lockServiceSpy.request.and.callFake( async (_name: string, callback: () => Promise) => callback(), ); + captureServiceSpy = jasmine.createSpyObj('OperationCaptureService', [ + 'getPendingCount', + ]); + captureServiceSpy.getPendingCount.and.returnValue(0); TestBed.configureTestingModule({ providers: [ OperationWriteFlushService, { provide: LockService, useValue: lockServiceSpy }, + { provide: OperationCaptureService, useValue: captureServiceSpy }, ], }); service = TestBed.inject(OperationWriteFlushService); @@ -102,6 +109,71 @@ describe('OperationWriteFlushService', () => { }); }); + describe('flushThenRunExclusive', () => { + it('should flush BEFORE acquiring the lock (calling flush inside the held lock deadlocks)', async () => { + const events: string[] = []; + spyOn(service, 'flushPendingWrites').and.callFake(async () => { + events.push('flush'); + }); + lockServiceSpy.request.and.callFake( + async (_name: string, callback: () => Promise) => { + events.push('lock-acquired'); + const result = await callback(); + events.push('lock-released'); + return result; + }, + ); + + await service.flushThenRunExclusive(async () => { + events.push('fn'); + }); + + expect(events).toEqual(['flush', 'lock-acquired', 'fn', 'lock-released']); + }); + + it('should return the value produced by fn', async () => { + spyOn(service, 'flushPendingWrites').and.resolveTo(); + + const result = await service.flushThenRunExclusive(async () => 42); + + expect(result).toBe(42); + }); + + it('should release, re-flush, and retry when a capture lands between flush and lock acquisition', async () => { + spyOn(service, 'flushPendingWrites').and.resolveTo(); + captureServiceSpy.getPendingCount.and.returnValues(1, 0); + const fn = jasmine.createSpy('fn').and.resolveTo('done'); + + const result = await service.flushThenRunExclusive(fn); + + expect(result).toBe('done'); + expect(service.flushPendingWrites).toHaveBeenCalledTimes(2); + expect(lockServiceSpy.request).toHaveBeenCalledTimes(2); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('should abort after bounded attempts under continuous dispatch activity', async () => { + spyOn(service, 'flushPendingWrites').and.resolveTo(); + captureServiceSpy.getPendingCount.and.returnValue(1); + const fn = jasmine.createSpy('fn'); + + await expectAsync(service.flushThenRunExclusive(fn)).toBeRejectedWithError( + /cutoff not reached/, + ); + expect(fn).not.toHaveBeenCalled(); + expect(service.flushPendingWrites).toHaveBeenCalledTimes(5); + }); + + it('should propagate fn rejections without retrying', async () => { + spyOn(service, 'flushPendingWrites').and.resolveTo(); + const testError = new Error('fn failed'); + const fn = jasmine.createSpy('fn').and.rejectWith(testError); + + await expectAsync(service.flushThenRunExclusive(fn)).toBeRejectedWith(testError); + expect(fn).toHaveBeenCalledTimes(1); + }); + }); + describe('FIFO ordering guarantee', () => { it('should ensure prior lock holders complete before flush resolves', async () => { // This test verifies the core guarantee: when flushPendingWrites resolves, diff --git a/src/app/op-log/sync/operation-write-flush.service.ts b/src/app/op-log/sync/operation-write-flush.service.ts index b36456380d..c3d6ab1bb4 100644 --- a/src/app/op-log/sync/operation-write-flush.service.ts +++ b/src/app/op-log/sync/operation-write-flush.service.ts @@ -39,6 +39,14 @@ export class OperationWriteFlushService { */ private readonly MAX_WAIT_TIME = 30000; + /** + * Maximum flush→lock→recheck attempts in flushThenRunExclusive before + * aborting. Bounds the retry loop so continuous dispatch activity (e.g. a + * runaway effect or 1Hz tracking ticks on a very slow device) cannot + * livelock the caller; the operation re-triggers on the next sync. + */ + private readonly MAX_CUTOFF_ATTEMPTS = 5; + /** * Polling interval to check queue size (ms). */ @@ -107,4 +115,43 @@ export class OperationWriteFlushService { `Initial pending: ${initialPendingCount}, Final pending: ${finalPendingCount}`, ); } + + /** + * Runs `fn` inside the operation-log lock with the capture pipeline drained — + * every action dispatched before the lock was acquired is durably written. + * + * flushPendingWrites() must NOT be called while holding the lock: its Phase 2 + * re-acquires the same non-reentrant lock and deadlocks until the acquisition + * timeout. So the flush runs BEFORE acquisition. A reducer action can still + * land in the tiny gap between the flush releasing the lock and our + * acquisition — its pending counter increments synchronously, before the + * persistence effect waits for the lock. Taking a snapshot/backup then would + * include that reducer state but precede its operation, so instead the lock + * is released, the writes are re-flushed, and the acquisition retried + * (bounded by MAX_CUTOFF_ATTEMPTS). + */ + async flushThenRunExclusive(fn: () => Promise): Promise { + for (let attempt = 0; attempt < this.MAX_CUTOFF_ATTEMPTS; attempt++) { + await this.flushPendingWrites(); + const outcome = await this.lockService.request( + LOCK_NAMES.OPERATION_LOG, + async (): Promise<{ retry: true } | { retry: false; value: T }> => { + if (this.captureService.getPendingCount() > 0) { + return { retry: true }; + } + return { retry: false, value: await fn() }; + }, + ); + if (!outcome.retry) { + return outcome.value; + } + OpLog.warn( + `OperationWriteFlushService: Capture landed between flush and lock acquisition; ` + + `retrying cutoff (attempt ${attempt + 1}/${this.MAX_CUTOFF_ATTEMPTS}).`, + ); + } + throw new Error( + `Operation write cutoff not reached after ${this.MAX_CUTOFF_ATTEMPTS} attempts — continuous dispatch activity. Try again.`, + ); + } } diff --git a/src/app/op-log/sync/server-migration.service.spec.ts b/src/app/op-log/sync/server-migration.service.spec.ts index 2eb1041845..03bbaa7adf 100644 --- a/src/app/op-log/sync/server-migration.service.spec.ts +++ b/src/app/op-log/sync/server-migration.service.spec.ts @@ -99,8 +99,16 @@ describe('ServerMigrationService', () => { ); writeFlushServiceSpy = jasmine.createSpyObj('OperationWriteFlushService', [ 'flushPendingWrites', + 'flushThenRunExclusive', ]); writeFlushServiceSpy.flushPendingWrites.and.resolveTo(); + // Mirror the real barrier semantics: flush, acquire the op-log lock, run fn. + writeFlushServiceSpy.flushThenRunExclusive.and.callFake( + async (fn: () => Promise) => { + await writeFlushServiceSpy.flushPendingWrites(); + return lockServiceSpy.request(LOCK_NAMES.OPERATION_LOG, fn); + }, + ); operationCaptureServiceSpy = jasmine.createSpyObj('OperationCaptureService', [ 'getPendingCount', ]); @@ -528,16 +536,9 @@ describe('ServerMigrationService', () => { ]); }); - it('should release, flush, and retry when an action lands before snapshot capture', async () => { - operationCaptureServiceSpy.getPendingCount.and.returnValues(1, 0); - - await service.handleServerMigration(defaultProvider); - - expect(writeFlushServiceSpy.flushPendingWrites).toHaveBeenCalledTimes(2); - expect(lockServiceSpy.request).toHaveBeenCalledTimes(2); - expect(stateSnapshotServiceSpy.getStateSnapshotAsync).toHaveBeenCalledTimes(1); - expect(opLogStoreSpy.append).toHaveBeenCalledTimes(1); - }); + // The release-flush-retry behavior when an action lands between flush and lock + // acquisition now lives in OperationWriteFlushService.flushThenRunExclusive — + // covered by operation-write-flush.service.spec.ts. describe('system-tag empty-state detection (tested via handleServerMigration)', () => { it('should identify system tags correctly', async () => { diff --git a/src/app/op-log/sync/server-migration.service.ts b/src/app/op-log/sync/server-migration.service.ts index 5e8330b3aa..b58985fdd1 100644 --- a/src/app/op-log/sync/server-migration.service.ts +++ b/src/app/op-log/sync/server-migration.service.ts @@ -25,10 +25,7 @@ import { CLIENT_ID_PROVIDER } from '../util/client-id.provider'; import { DialogServerMigrationConfirmComponent } from './dialog-server-migration-confirm/dialog-server-migration-confirm.component'; import { hasMeaningfulStateData } from '../validation/has-meaningful-state-data.util'; import { AppDataComplete, MODEL_CONFIGS } from '../model/model-config'; -import { LockService } from './lock.service'; import { OperationWriteFlushService } from './operation-write-flush.service'; -import { LOCK_NAMES } from '../core/operation-log.const'; -import { OperationCaptureService } from '../capture/operation-capture.service'; const MEANINGFUL_ENTITY_STATE_KEYS = new Set(['task', 'project', 'tag', 'note']); @@ -90,9 +87,7 @@ export class ServerMigrationService { private clientIdProvider = inject(CLIENT_ID_PROVIDER); private _matDialog = inject(MatDialog); private _userInputWaitState = inject(UserInputWaitStateService); - private lockService = inject(LockService); private writeFlushService = inject(OperationWriteFlushService); - private operationCaptureService = inject(OperationCaptureService); /** * Checks if we're connecting to a new/empty server and handles migration if needed. @@ -209,125 +204,113 @@ export class ServerMigrationService { // construction, and append behind the same mutation barrier. This makes the // full-state operation's local seq an exact cutoff: every earlier op is in the // snapshot, and any action captured while this runs is appended afterwards. - let retryForPendingCapture: boolean; - do { - retryForPendingCapture = false; - await this.writeFlushService.flushPendingWrites(); - await this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => { - // A reducer action can land in the tiny gap between flush() releasing - // this lock and our acquisition. Its pending counter increments - // synchronously, before the persistence effect waits for the lock. Do - // not take a snapshot that includes that reducer state but precedes its - // operation; release, drain it, and retry the cutoff instead. - if (this.operationCaptureService.getPendingCount() > 0) { - retryForPendingCapture = true; - return; - } + // flushThenRunExclusive owns the flush→lock→recheck retry loop (bounded, so + // continuous dispatch cannot livelock the migration; it re-triggers on the + // next sync). + await this.writeFlushService.flushThenRunExclusive(async () => { + // Get current full state from NgRx store (async to include archives from IndexedDB) + // Cast to Record for validation compatibility + let currentState: Record = + (await this.stateSnapshotService.getStateSnapshotAsync()) as unknown as Record< + string, + unknown + >; - // Get current full state from NgRx store (async to include archives from IndexedDB) - // Cast to Record for validation compatibility - let currentState: Record = - (await this.stateSnapshotService.getStateSnapshotAsync()) as unknown as Record< - string, - unknown - >; - - // Skip if local state is effectively empty - if (!hasServerMigrationStateData(currentState)) { - OpLog.warn( - 'ServerMigrationService: Skipping SYNC_IMPORT - local state is empty.', - ); - return; - } - - // Validate and repair state before creating SYNC_IMPORT - // This prevents corrupted state (e.g., orphaned menuTree references) from - // propagating to other clients via the full state import. - const validationResult = - await this.validateStateService.validateAndRepair(currentState); - - // If state is invalid and couldn't be repaired, abort - don't propagate corruption - if (!validationResult.isValid) { - OpLog.err( - 'ServerMigrationService: Cannot create SYNC_IMPORT - state validation failed.', - validationResult.error || validationResult.crossModelError, - ); - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.SERVER_MIGRATION_VALIDATION_FAILED, - }); - return; - } - - // If state was repaired, use the repaired version - if (validationResult.repairedState) { - OpLog.warn( - 'ServerMigrationService: State repaired before creating SYNC_IMPORT', - validationResult.repairSummary, - ); - currentState = validationResult.repairedState; - - // Also update NgRx store with repaired state so local client is consistent - this.store.dispatch( - loadAllData({ - appDataComplete: validationResult.repairedState as AppDataComplete, - }), - ); - } - - // Get client ID - const clientId = await this.clientIdProvider.loadClientId(); - if (!clientId) { - OpLog.err( - 'ServerMigrationService: Cannot create SYNC_IMPORT - no client ID available.', - ); - return; - } - - // Build vector clock by merging ALL local operation clocks. - // This ensures the SYNC_IMPORT's clock dominates all pre-import ops, - // so when SyncImportFilterService compares them, all prior ops are - // LESS_THAN (not CONCURRENT) and can be properly filtered. - const allLocalOps = await this.opLogStore.getOpsAfterSeq(0); - let mergedClock = await this.vectorClockService.getCurrentVectorClock(); - for (const entry of allLocalOps) { - mergedClock = mergeVectorClocks(mergedClock, entry.op.vectorClock); - } - const newClock = limitVectorClockSize( - incrementVectorClock(mergedClock, clientId), - clientId, + // Skip if local state is effectively empty + if (!hasServerMigrationStateData(currentState)) { + OpLog.warn( + 'ServerMigrationService: Skipping SYNC_IMPORT - local state is empty.', ); + return; + } - OpLog.normal( - `ServerMigrationService: Merged ${allLocalOps.length} local op clocks into SYNC_IMPORT vector clock.`, + // Validate and repair state before creating SYNC_IMPORT + // This prevents corrupted state (e.g., orphaned menuTree references) from + // propagating to other clients via the full state import. + const validationResult = + await this.validateStateService.validateAndRepair(currentState); + + // If state is invalid and couldn't be repaired, abort - don't propagate corruption + if (!validationResult.isValid) { + OpLog.err( + 'ServerMigrationService: Cannot create SYNC_IMPORT - state validation failed.', + validationResult.error || validationResult.crossModelError, ); + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.SERVER_MIGRATION_VALIDATION_FAILED, + }); + return; + } - // Create SYNC_IMPORT operation with full state - // NOTE: Use raw state directly (not wrapped in appDataComplete). - // The snapshot endpoint expects raw state, and the hydrator handles - // both formats on extraction. - const op: Operation = { - id: uuidv7(), - actionType: ActionType.LOAD_ALL_DATA, - opType: OpType.SyncImport, - entityType: 'ALL', - payload: currentState, - clientId, - vectorClock: newClock, - timestamp: Date.now(), - schemaVersion: CURRENT_SCHEMA_VERSION, - syncImportReason: options?.syncImportReason ?? 'SERVER_MIGRATION', - }; - - // Append to operation log - will be uploaded via snapshot endpoint - await this.opLogStore.append(op, 'local'); - - OpLog.normal( - 'ServerMigrationService: Created SYNC_IMPORT operation for server migration. ' + - 'Will be uploaded immediately via follow-up upload.', + // If state was repaired, use the repaired version + if (validationResult.repairedState) { + OpLog.warn( + 'ServerMigrationService: State repaired before creating SYNC_IMPORT', + validationResult.repairSummary, ); - }); - } while (retryForPendingCapture); + currentState = validationResult.repairedState; + + // Also update NgRx store with repaired state so local client is consistent + this.store.dispatch( + loadAllData({ + appDataComplete: validationResult.repairedState as AppDataComplete, + }), + ); + } + + // Get client ID + const clientId = await this.clientIdProvider.loadClientId(); + if (!clientId) { + OpLog.err( + 'ServerMigrationService: Cannot create SYNC_IMPORT - no client ID available.', + ); + return; + } + + // Build vector clock by merging ALL local operation clocks. + // This ensures the SYNC_IMPORT's clock dominates all pre-import ops, + // so when SyncImportFilterService compares them, all prior ops are + // LESS_THAN (not CONCURRENT) and can be properly filtered. + const allLocalOps = await this.opLogStore.getOpsAfterSeq(0); + let mergedClock = await this.vectorClockService.getCurrentVectorClock(); + for (const entry of allLocalOps) { + mergedClock = mergeVectorClocks(mergedClock, entry.op.vectorClock); + } + const newClock = limitVectorClockSize( + incrementVectorClock(mergedClock, clientId), + clientId, + ); + + OpLog.normal( + `ServerMigrationService: Merged ${allLocalOps.length} local op clocks into SYNC_IMPORT vector clock.`, + ); + + // Create SYNC_IMPORT operation with full state + // NOTE: Use raw state directly (not wrapped in appDataComplete). + // The snapshot endpoint expects raw state, and the hydrator handles + // both formats on extraction. + const op: Operation = { + id: uuidv7(), + actionType: ActionType.LOAD_ALL_DATA, + opType: OpType.SyncImport, + entityType: 'ALL', + payload: currentState, + clientId, + vectorClock: newClock, + timestamp: Date.now(), + schemaVersion: CURRENT_SCHEMA_VERSION, + syncImportReason: options?.syncImportReason ?? 'SERVER_MIGRATION', + }; + + // Append to operation log - will be uploaded via snapshot endpoint + await this.opLogStore.append(op, 'local'); + + OpLog.normal( + 'ServerMigrationService: Created SYNC_IMPORT operation for server migration. ' + + 'Will be uploaded immediately via follow-up upload.', + ); + }); } /** diff --git a/src/app/op-log/testing/integration/schema-version-sync.integration.spec.ts b/src/app/op-log/testing/integration/schema-version-sync.integration.spec.ts index bb3aa32f86..bba9abd9e3 100644 --- a/src/app/op-log/testing/integration/schema-version-sync.integration.spec.ts +++ b/src/app/op-log/testing/integration/schema-version-sync.integration.spec.ts @@ -9,7 +9,6 @@ import { resetTestUuidCounter } from './helpers/test-client.helper'; import { MockSyncServer } from './helpers/mock-sync-server.helper'; import { SimulatedClient } from './helpers/simulated-client.helper'; import { createMinimalTaskPayload } from './helpers/operation-factory.helper'; -import { MAX_VERSION_SKIP } from '@sp/shared-schema'; /** * Integration tests for Schema Version Sync. @@ -175,11 +174,6 @@ describe('Schema Version Sync Integration', () => { expect(CURRENT_SCHEMA_VERSION).toBeGreaterThanOrEqual(1); }); - it('should define MAX_VERSION_SKIP as a positive number', () => { - expect(typeof MAX_VERSION_SKIP).toBe('number'); - expect(MAX_VERSION_SKIP).toBeGreaterThan(0); - }); - it('should not need migration for state at current version', () => { const cache = { state: {},