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: {},