From f9610530c90de3ca303399a302c803b4e3633db9 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Sat, 11 Jul 2026 14:49:08 +0200 Subject: [PATCH] fix(sync): converge rebuild capture races, unwedge archive recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remediate the findings of the 5-agent necessity review: - USE_REMOTE: a local capture racing the locked rebuild now throws a typed CaptureRacedRebuildError, and forceDownloadRemoteState retries phase 2 in-call (bounded, 3 attempts) through the existing crash-resume branch — raced ops fold into preservedLocalOps and the already-downloaded history is reused (WS downloads and immediate uploads stay gated by the marker, so no re-download is needed). Previously every attempt aborted while e.g. time tracking dispatched continuously, re-downloading the full history per sync trigger and churning the Undo snack (now shown on final failure only). - Hydrator archive retry: pass skipDeferredLocalActions and drain explicitly with a caught error. A drain throw from the coordinator's finally used to mask the archive result and escalate out of hydrateStore() into attemptRecovery(), which can import stale legacy data over a correctly hydrated store. - Incomplete-remote gate: run one in-session archive-only retry when the only blockers are quarantined failed/archive_pending ops, so a transient archive failure self-heals on the next sync instead of wedging until app restart. Never attempted while the rebuild marker is set or reducer-uncommitted pending rows exist. - Boot recovery: StartupService surfaces the pre-replace backup's persistent restore snack when a stranded raw_rebuild_incomplete marker is found, covering users who boot offline or disable sync after finding the app "emptied" by a mid-rebuild crash. - Snack correctness: latch the USE_REMOTE newer-schema warning once per session; guard _notifyBlockedOp and the LWW apply-failure snack with hasPendingPersistentAction() so they cannot destroy a visible Undo; drop the useless "Update app" action for below-minimum data. - Strings: MIGRATION_FAILED / VERSION_UNSUPPORTED now describe the blocking semantics instead of the removed skip behavior. - Store: getPendingRemoteOps excludes rejected rows (parity with getFailedRemoteOps) so a rejected-but-pending row cannot trip the sync gate for a session. - Server: compute the upload request fingerprint eagerly after the rate-limit gate — identical cost to the lazy closure in every path, minus the memoization machinery; laziness remains in the snapshot handler where it skips hashing multi-MB states. op-log suite 3004/3004, super-sync-server 831/831, checkFile clean on all touched files. Nine regression tests pin the new behavior. --- .../src/sync/sync.routes.ops-handler.ts | 40 +- src/app/core/startup/startup.service.ts | 26 ++ src/app/op-log/core/errors/sync-errors.ts | 17 + ...ion-log-hydrator.retry.integration.spec.ts | 1 + .../operation-log-hydrator.service.spec.ts | 36 +- .../operation-log-hydrator.service.ts | 98 +++-- .../operation-log-store.service.spec.ts | 13 + .../operation-log-store.service.ts | 6 +- .../sync/conflict-resolution.service.spec.ts | 6 +- .../sync/conflict-resolution.service.ts | 22 +- .../sync/operation-log-sync.service.spec.ts | 97 +++++ .../op-log/sync/operation-log-sync.service.ts | 395 +++++++++++------- .../remote-ops-processing.service.spec.ts | 20 +- .../sync/remote-ops-processing.service.ts | 31 +- .../sync/sync-import-conflict-gate.service.ts | 10 +- .../migration-handling.integration.spec.ts | 6 +- src/assets/i18n/en.json | 4 +- 17 files changed, 584 insertions(+), 244 deletions(-) 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 29f4e15c3a..568ad1c689 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,15 +87,6 @@ export const uploadOpsHandler = async ( } const { ops, clientId, lastKnownServerSeq, requestId } = parseResult.data; - // 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( @@ -118,14 +109,25 @@ export const uploadOpsHandler = async ( }); } + // Compute the request fingerprint AFTER the rate-limit gate (a rate-limited + // client must not burn CPU on it) and BEFORE any processing: uploadOps and + // the quota prevalidation mutate the parsed ops (e.g. vector-clock + // sanitizing/pruning), so hashing later would never match a retry's + // pre-processing hash. Eager is as cheap as lazy here — every non-limited + // requestId-bearing request needs the hash exactly once (either the dedup + // check below or the post-upload cache write). + const requestFingerprint = requestId + ? createOpsRequestFingerprint(clientId, ops as unknown as Operation[]) + : undefined; + // Check for duplicate request (client retry) BEFORE quota check // This ensures retries after successful uploads don't fail with 413 // if the original upload pushed the user over quota - if (requestId) { + if (requestId && requestFingerprint) { const cachedResults = syncService.checkOpsRequestDedup( userId, requestId, - getRequestFingerprint, + () => requestFingerprint, ); if (cachedResults) { Logger.info(`[user:${userId}] Returning cached results for request ${requestId}`); @@ -176,14 +178,6 @@ 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 () => { @@ -242,14 +236,10 @@ export const uploadOpsHandler = async ( // batch to it, so a single match means the transaction failed. #8332 if ( requestId && + requestFingerprint && !results.some((r) => r.errorCode === SYNC_ERROR_CODES.INTERNAL_ERROR) ) { - syncService.cacheOpsRequestResults( - userId, - requestId, - results, - getRequestFingerprint(), - ); + syncService.cacheOpsRequestResults(userId, requestId, results, requestFingerprint); } const accepted = results.filter((r) => r.accepted).length; diff --git a/src/app/core/startup/startup.service.ts b/src/app/core/startup/startup.service.ts index 2fe1bf006d..8623841fda 100644 --- a/src/app/core/startup/startup.service.ts +++ b/src/app/core/startup/startup.service.ts @@ -14,6 +14,7 @@ import { IS_ELECTRON } from '../../app.constants'; import { Log } from '../log'; import { T } from '../../t.const'; import { OperationLogStoreService } from '../../op-log/persistence/operation-log-store.service'; +import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service'; import { LegacyPfDbService } from '../persistence/legacy-pf-db.service'; import { BannerId } from '../banner/banner.model'; import { isOnline$ } from '../../util/is-online'; @@ -169,6 +170,9 @@ export class StartupService { this._ratePromptService.init(); await this._initPlugins(); + // Last in the deferred body: the snack it may open is persistent and the + // single snack slot must not be reclaimed by the productivity tip above. + await this._offerInterruptedRebuildRecoveryIfNeeded(); }, DEFERRED_INIT_DELAY_MS); if (IS_ELECTRON) { @@ -258,6 +262,28 @@ export class StartupService { } } + /** + * An interrupted USE_REMOTE rebuild leaves the user booting into the rebuild + * baseline instead of their data. Sync (when it runs) resumes the rebuild by + * itself — but when it cannot (offline, or the user disabled sync after + * finding the app "emptied" by the crash), the pre-replace backup would have + * no visible entry point. Surfaces the persistent restore snack in that case. + */ + private async _offerInterruptedRebuildRecoveryIfNeeded(): Promise { + try { + if (await this._opLogStore.isRawRebuildIncomplete()) { + await this._injector + .get(OperationLogSyncService) + .offerInterruptedRebuildRecovery(); + } + } catch (err) { + Log.err({ + stage: 'interrupted-rebuild-recovery-check', + error: (err as Error)?.message, + }); + } + } + private async _checkIsSingleInstance(): Promise { const channel = new BroadcastChannel('superProductivityTab'); let isAnotherInstanceActive = false; diff --git a/src/app/op-log/core/errors/sync-errors.ts b/src/app/op-log/core/errors/sync-errors.ts index e145490b17..2df0ad44c3 100644 --- a/src/app/op-log/core/errors/sync-errors.ts +++ b/src/app/op-log/core/errors/sync-errors.ts @@ -122,6 +122,23 @@ export class PermanentDeferredWriteError extends Error { override name = 'PermanentDeferredWriteError'; } +/** + * A local action was captured while a USE_REMOTE rebuild held the op-log lock, + * after the destructive replacement committed. The attempt must abort — the + * raced action's reducer ran against live state the replay rewrites, so + * completing could let a later snapshot cover an op whose live effect is + * missing. The raced ops are preserved and re-applied by the retry/resume. + */ +export class CaptureRacedRebuildError extends Error { + override name = 'CaptureRacedRebuildError'; + + constructor() { + super( + 'USE_REMOTE incomplete: a local change arrived during the rebuild and will be restored on retry.', + ); + } +} + /** Previously downloaded operations have not completed reducer/archive recovery. */ export class IncompleteRemoteOperationsError extends Error { override name = 'IncompleteRemoteOperationsError'; diff --git a/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts b/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts index 1b42d0afd5..32ff621cdd 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts @@ -135,6 +135,7 @@ describe('OperationLogHydratorService retryFailedRemoteOps (integration, real st // double-apply additive reducers. expect(applier.applyOperations.calls.argsFor(0)[1]).toEqual({ skipReducerDispatch: true, + skipDeferredLocalActions: true, }); }); 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 a207c9d916..abd3dfb501 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 @@ -31,6 +31,7 @@ import { MAX_VECTOR_CLOCK_SIZE } from '../core/operation-log.const'; import { IndexedDBOpenError } from '../core/errors/indexed-db-open.error'; import { IDB_OPEN_ERROR_RELOAD_KEY } from './operation-log-hydrator.service'; import { SyncProviderId } from '../sync-providers/provider.const'; +import { OperationLogEffects } from '../capture/operation-log.effects'; describe('OperationLogHydratorService', () => { let service: OperationLogHydratorService; @@ -44,6 +45,7 @@ describe('OperationLogHydratorService', () => { let mockRepairOperationService: jasmine.SpyObj; let mockVectorClockService: jasmine.SpyObj; let mockOperationApplierService: jasmine.SpyObj; + let mockOperationLogEffects: jasmine.SpyObj; let mockHydrationStateService: jasmine.SpyObj; let mockSnapshotService: jasmine.SpyObj; let mockRecoveryService: jasmine.SpyObj; @@ -139,6 +141,10 @@ describe('OperationLogHydratorService', () => { mockOperationApplierService = jasmine.createSpyObj('OperationApplierService', [ 'applyOperations', ]); + mockOperationLogEffects = jasmine.createSpyObj('OperationLogEffects', [ + 'processDeferredActions', + ]); + mockOperationLogEffects.processDeferredActions.and.resolveTo(); mockHydrationStateService = jasmine.createSpyObj('HydrationStateService', [ 'startApplyingRemoteOps', 'endApplyingRemoteOps', @@ -216,6 +222,7 @@ describe('OperationLogHydratorService', () => { { provide: RepairOperationService, useValue: mockRepairOperationService }, { provide: VectorClockService, useValue: mockVectorClockService }, { provide: OperationApplierService, useValue: mockOperationApplierService }, + { provide: OperationLogEffects, useValue: mockOperationLogEffects }, { provide: HydrationStateService, useValue: mockHydrationStateService }, { provide: OperationLogSnapshotService, useValue: mockSnapshotService }, { provide: OperationLogRecoveryService, useValue: mockRecoveryService }, @@ -478,7 +485,10 @@ describe('OperationLogHydratorService', () => { const [retriedOps, retryOptions] = mockOperationApplierService.applyOperations.calls.argsFor(0); expect(retriedOps.map((o: Operation) => o.id)).toEqual(['op-failed']); - expect(retryOptions).toEqual({ skipReducerDispatch: true }); + expect(retryOptions).toEqual({ + skipReducerDispatch: true, + skipDeferredLocalActions: true, + }); expect(mockOpLogStore.markApplied).toHaveBeenCalledWith([6]); }); @@ -1457,7 +1467,10 @@ describe('OperationLogHydratorService', () => { await service.retryFailedRemoteOps(); const options = mockOperationApplierService.applyOperations.calls.argsFor(0)[1]; - expect(options).toEqual({ skipReducerDispatch: true }); + expect(options).toEqual({ + skipReducerDispatch: true, + skipDeferredLocalActions: true, + }); }); it('should apply failed ops in ascending seq order regardless of store order', async () => { @@ -1515,6 +1528,25 @@ describe('OperationLogHydratorService', () => { expect(mockOpLogStore.markApplied).not.toHaveBeenCalled(); }); + it('should not escalate a deferred-drain failure into hydration recovery', async () => { + // A drain throw here used to propagate out of hydrateStore() into + // attemptRecovery(), which can import stale legacy data over a correctly + // hydrated store. The failure is logged; buffered actions stay queued + // for the next drain point (e.g. the pre-sync flush). + mockOperationLogEffects.processDeferredActions.and.rejectWith( + new Error('drain failed'), + ); + mockOpLogStore.getFailedRemoteOps.and.resolveTo([failedEntry(40, 'op-a')]); + mockOperationApplierService.applyOperations.and.callFake((ops: Operation[]) => + Promise.resolve({ appliedOps: ops }), + ); + + await expectAsync(service.retryFailedRemoteOps()).toBeResolved(); + + // Bookkeeping completed despite the failed drain. + expect(mockOpLogStore.markApplied).toHaveBeenCalledWith([40]); + }); + it('should do nothing when there are no failed ops', async () => { mockOpLogStore.getFailedRemoteOps.and.returnValue(Promise.resolve([])); 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 108d0f05a6..27386e7fd0 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.ts @@ -1,6 +1,7 @@ -import { inject, Injectable } from '@angular/core'; +import { inject, Injectable, Injector } from '@angular/core'; import { Store } from '@ngrx/store'; import { OperationLogStoreService } from './operation-log-store.service'; +import { processDeferredActions } from '../sync/process-deferred-actions-flush.util'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; import { OperationLogMigrationService } from './operation-log-migration.service'; import { @@ -54,6 +55,7 @@ export class OperationLogHydratorService { private vectorClockService = inject(VectorClockService); private operationApplierService = inject(OperationApplierService); private hydrationStateService = inject(HydrationStateService); + private injector = inject(Injector); private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); // Extracted services @@ -628,45 +630,67 @@ export class OperationLogHydratorService { // outstanding archive side effects: re-dispatching would double-apply // additive reducers (syncTimeSpent, increaseSimpleCounterCounterToday) // on every retry attempt. - const result = await this.operationApplierService.applyOperations(opsToApply, { - skipReducerDispatch: true, - }); + try { + const result = await this.operationApplierService.applyOperations(opsToApply, { + skipReducerDispatch: true, + // The drain runs in the finally below with its own error boundary. Left + // to the applier's finally, a drain throw would mask the archive result + // (markFailed below never runs) and escalate out of hydrateStore() into + // attemptRecovery(), which can import stale legacy data over a + // correctly hydrated store. + skipDeferredLocalActions: true, + }); - // Mark successfully applied ops. - const appliedSeqs = result.appliedOps - .map((op) => opIdToSeq.get(op.id)) - .filter((seq): seq is number => seq !== undefined); - if (appliedSeqs.length > 0) { - // 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); - await this.opLogStore.markApplied(appliedSeqs); - OpLog.normal( - `OperationLogHydratorService: Successfully retried ${appliedSeqs.length} failed ops`, - ); - } + // Mark successfully applied ops. + const appliedSeqs = result.appliedOps + .map((op) => opIdToSeq.get(op.id)) + .filter((seq): seq is number => seq !== undefined); + if (appliedSeqs.length > 0) { + // 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); + await this.opLogStore.markApplied(appliedSeqs); + OpLog.normal( + `OperationLogHydratorService: Successfully retried ${appliedSeqs.length} failed ops`, + ); + } - // On a partial failure the batch applier stops at the first archive error. - // Charge only that attempted operation: successors remain archive-pending - // without consuming retry budget and will run after the blocker succeeds. - // A persistent blocker stays failed so ordinary sync remains safely paused. - if (result.failedOp) { - const failedOpIds = [result.failedOp.op.id]; + // On a partial failure the batch applier stops at the first archive error. + // Charge only that attempted operation: successors remain archive-pending + // without consuming retry budget and will run after the blocker succeeds. + // A persistent blocker stays failed so ordinary sync remains safely paused. + if (result.failedOp) { + const failedOpIds = [result.failedOp.op.id]; - OpLog.warn( - `OperationLogHydratorService: Failed to retry op ${result.failedOp.op.id}`, - result.failedOp.error, - ); - // Keep archive failure visible to the sync safety gate. A retry cap that - // rejects it would hide incomplete downloaded work and allow false IN_SYNC. - await this.opLogStore.markFailed(failedOpIds); - OpLog.warn( - 'OperationLogHydratorService: Archive operation still failing after retry', - ); + OpLog.warn( + `OperationLogHydratorService: Failed to retry op ${result.failedOp.op.id}`, + result.failedOp.error, + ); + // Keep archive failure visible to the sync safety gate. A retry cap that + // rejects it would hide incomplete downloaded work and allow false IN_SYNC. + await this.opLogStore.markFailed(failedOpIds); + OpLog.warn( + 'OperationLogHydratorService: Archive operation still failing after retry', + ); + } + } finally { + // Local actions captured while the retry held the remote-apply window + // open. Runs after mergeRemoteOpClocks so their clocks dominate the + // retried remote ops (#7700). A failed drain keeps the actions buffered + // for the next drain point (e.g. the pre-sync flush) — never escalate + // it into hydration recovery. + try { + await processDeferredActions(this.injector, false); + } catch (drainError) { + OpLog.err( + 'OperationLogHydratorService: Deferred-action drain failed after archive retry; actions stay buffered.', + { name: (drainError as Error | undefined)?.name }, + ); + } } } 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 70280af355..1e47f8d613 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 @@ -1558,6 +1558,19 @@ describe('OperationLogStoreService', () => { expect(pending.length).toBe(0); }); + + it('should exclude rejected ops (parity with getFailedRemoteOps)', async () => { + // A rejected-but-still-pending row must not trip the incomplete-remote + // sync gate: nothing will ever apply it, so counting it would wedge sync + // for the whole session. + const op = createTestOperation({ entityId: 'rejected-pending' }); + await service.append(op, 'remote', { pendingApply: true }); + await service.markRejected([op.id]); + + const pending = await service.getPendingRemoteOps(); + + expect(pending.length).toBe(0); + }); }); describe('markFailed', () => { 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 a94ed39154..750de52454 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -765,8 +765,10 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort entry.source === 'remote' && entry.applicationStatus === 'pending', ); } - // Decode compact operations for backwards compatibility - return storedEntries.map(decodeStoredEntry); + // Exclude rejected ops (mirrors getFailedRemoteOps): a rejected-but-still- + // pending row must not trip the incomplete-remote sync gate — nothing will + // ever apply it, so it would wedge sync for the whole session. + return storedEntries.filter((e) => !e.rejectedAt).map(decodeStoredEntry); } async hasOp(id: string): Promise { diff --git a/src/app/op-log/sync/conflict-resolution.service.spec.ts b/src/app/op-log/sync/conflict-resolution.service.spec.ts index d618cfdeed..4ec9381732 100644 --- a/src/app/op-log/sync/conflict-resolution.service.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.service.spec.ts @@ -64,7 +64,11 @@ describe('ConflictResolutionService', () => { 'mergeRemoteOpClocks', ]); mockOpLogStore.mergeRemoteOpClocks.and.resolveTo(undefined); - mockSnackService = jasmine.createSpyObj('SnackService', ['open']); + mockSnackService = jasmine.createSpyObj('SnackService', [ + 'open', + 'hasPendingPersistentAction', + ]); + mockSnackService.hasPendingPersistentAction.and.returnValue(false); mockValidateStateService = jasmine.createSpyObj('ValidateStateService', [ 'validateAndRepairCurrentState', ]); diff --git a/src/app/op-log/sync/conflict-resolution.service.ts b/src/app/op-log/sync/conflict-resolution.service.ts index f8538fa034..5c6e8ee11a 100644 --- a/src/app/op-log/sync/conflict-resolution.service.ts +++ b/src/app/op-log/sync/conflict-resolution.service.ts @@ -466,14 +466,20 @@ export class ConflictResolutionService { ); await this.opLogStore.markFailed(failedOpIds); - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.CONFLICT_RESOLUTION_FAILED, - actionStr: T.PS.RELOAD, - actionFn: (): void => { - window.location.reload(); - }, - }); + // Never replace a visible persistent recovery action (e.g. the + // USE_REMOTE Undo — the only entry point to the pre-replace backup). + // The IncompleteRemoteOperationsError thrown below still flips the + // sync status to ERROR via the wrapper's (equally guarded) handler. + if (!this.snackService.hasPendingPersistentAction()) { + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.CONFLICT_RESOLUTION_FAILED, + actionStr: T.PS.RELOAD, + actionFn: (): void => { + window.location.reload(); + }, + }); + } // FIX #6571: Throw on apply failure (parity with applyNonConflictingOps). // Previously, apply failures during LWW resolution were logged but not 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 0ac19981b6..35b00d15b9 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 @@ -2,6 +2,7 @@ import { TestBed } from '@angular/core/testing'; import { OperationLogSyncService } from './operation-log-sync.service'; import { FILE_BASED_SYNC_CONSTANTS } from '../sync-providers/file-based/file-based-sync.types'; import { SchemaMigrationService } from '../persistence/schema-migration.service'; +import { OperationLogHydratorService } from '../persistence/operation-log-hydrator.service'; import { SnackService } from '../../core/snack/snack.service'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { VectorClockService } from './vector-clock.service'; @@ -67,6 +68,7 @@ describe('OperationLogSyncService', () => { let lockServiceSpy: jasmine.SpyObj; let operationApplierSpy: jasmine.SpyObj; let operationLogEffectsSpy: jasmine.SpyObj; + let hydratorServiceSpy: jasmine.SpyObj; const createProviderSetupEntry = (): OperationLogEntry => ({ seq: 1, @@ -225,6 +227,10 @@ describe('OperationLogSyncService', () => { 'processDeferredActions', ]); operationLogEffectsSpy.processDeferredActions.and.resolveTo(); + hydratorServiceSpy = jasmine.createSpyObj('OperationLogHydratorService', [ + 'retryFailedRemoteOps', + ]); + hydratorServiceSpy.retryFailedRemoteOps.and.resolveTo(); TestBed.configureTestingModule({ providers: [ @@ -247,6 +253,7 @@ describe('OperationLogSyncService', () => { useValue: operationApplierSpy, }, { provide: OperationLogEffects, useValue: operationLogEffectsSpy }, + { provide: OperationLogHydratorService, useValue: hydratorServiceSpy }, { provide: ConflictResolutionService, useValue: jasmine.createSpyObj('ConflictResolutionService', [ @@ -398,6 +405,16 @@ describe('OperationLogSyncService', () => { expect(uploadServiceSpy.uploadPendingOps).not.toHaveBeenCalled(); }); + it('should not attempt the in-session archive retry while a raw rebuild is incomplete', async () => { + opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(true); + + await expectAsync( + service.uploadPendingOps({} as OperationSyncCapable), + ).toBeRejectedWithError(IncompleteRemoteOperationsError); + + expect(hydratorServiceSpy.retryFailedRemoteOps).not.toHaveBeenCalled(); + }); + it('should return localWinOpsCreated: 0 when no piggybacked ops', async () => { opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([])); uploadServiceSpy.uploadPendingOps.and.returnValue( @@ -1000,9 +1017,31 @@ describe('OperationLogSyncService', () => { service.downloadRemoteOps({} as OperationSyncCapable), ).toBeRejected(); + // The one in-session repair attempt ran but couldn't clear the gate. + expect(hydratorServiceSpy.retryFailedRemoteOps).toHaveBeenCalledTimes(1); expect(downloadServiceSpy.downloadRemoteOps).not.toHaveBeenCalled(); }); + it('should proceed when the in-session archive retry clears the incomplete-remote gate', async () => { + // Transient archive failure: quarantined at gate read, gone on re-check. + opLogStoreSpy.getFailedRemoteOps.and.returnValues( + Promise.resolve([{ applicationStatus: 'failed' } as OperationLogEntry]), + Promise.resolve([]), + ); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + }); + + await service.downloadRemoteOps({} as OperationSyncCapable); + + expect(hydratorServiceSpy.retryFailedRemoteOps).toHaveBeenCalledTimes(1); + expect(downloadServiceSpy.downloadRemoteOps).toHaveBeenCalled(); + }); + 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. @@ -3051,6 +3090,64 @@ describe('OperationLogSyncService', () => { expect(opLogStoreSpy.clearRawRebuildIncomplete).not.toHaveBeenCalled(); }); + it('should retry the rebuild in-call on a capture race and converge without re-downloading', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 50, + }); + // Attempt 1 trips the completion assert (e.g. a tracking tick landed in + // an unprotected gap); the in-call retry runs clean. + writeFlushServiceSpy.hasPendingWrites.and.returnValues(true, false, false); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as any; + + await service.forceDownloadRemoteState(mockProvider); + + // One network download, two local rebuild attempts. + expect(downloadServiceSpy.downloadRemoteOps).toHaveBeenCalledTimes(1); + expect(opLogStoreSpy.runRemoteStateReplacement).toHaveBeenCalledTimes(2); + // The retry re-enters through the crash-resume branch: the FIRST + // attempt's pre-replace backup is kept, never re-captured over. + expect(backupServiceSpy.captureImportBackup).toHaveBeenCalledTimes(1); + expect(opLogStoreSpy.loadImportBackup).toHaveBeenCalled(); + expect(opLogStoreSpy.clearRawRebuildIncomplete).toHaveBeenCalled(); + }); + + it('should warn only once per session when the remote history requires a newer app', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [{ ...makeRemoteOp(), schemaVersion: 9999 }], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as any; + + await expectAsync( + service.forceDownloadRemoteState(mockProvider), + ).toBeRejectedWithError(/newer schema version/); + await expectAsync( + service.forceDownloadRemoteState(mockProvider), + ).toBeRejectedWithError(/newer schema version/); + + const versionSnackCount = snackServiceSpy.open.calls + .allArgs() + .filter( + ([cfg]) => typeof cfg !== 'string' && cfg.msg === T.F.SYNC.S.VERSION_TOO_OLD, + ).length; + expect(versionSnackCount).toBe(1); + }); + 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 8e1416735d..28738c9ceb 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -22,6 +22,7 @@ import { OperationLogDownloadService } from './operation-log-download.service'; import { SnackService } from '../../core/snack/snack.service'; import { T } from '../../t.const'; import { + CaptureRacedRebuildError, IncompleteRemoteOperationsError, LocalDataConflictError, } from '../core/errors/sync-errors'; @@ -47,6 +48,7 @@ import { SchemaMigrationService, getOperationSchemaVersion, } from '../persistence/schema-migration.service'; +import { OperationLogHydratorService } from '../persistence/operation-log-hydrator.service'; import { SyncProviderManager } from '../sync-providers/provider-manager.service'; import { getDefaultMainModelData, MODEL_CONFIGS } from '../model/model-config'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; @@ -162,6 +164,13 @@ export class OperationLogSyncService { private operationApplier = inject(OperationApplierService); private injector = inject(Injector); + /** + * Once-per-session latch for the USE_REMOTE newer-schema snack: the block + * persists until an app update, and every auto/WS-triggered resume attempt + * re-hits the preflight — without the latch the snack re-fires each time. + */ + private _hasWarnedRebuildVersionBlockThisSession = false; + /** * Checks if this client is "wholly fresh" - meaning it has never synced before * and has no local operation history. A fresh client accepting remote data @@ -1427,96 +1436,160 @@ export class OperationLogSyncService { let replacementCommitted = false; let backupSavedAt: number | undefined; let preservedLocalOps: Operation[] = []; - try { - // 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; - const marker = await this.opLogStore.loadRawRebuildIncomplete(); - const liveLocalOps = (await this.opLogStore.getUnsynced()).map( - (entry) => entry.op, - ); - preservedLocalOps = this._mergeOperationsById( - marker?.preservedLocalOps ?? [], - liveLocalOps, - ); - } else { - try { - savedAt = await this.backupService.captureImportBackup(); + // A capture racing the rebuild aborts the attempt (CaptureRacedRebuildError + // from the asserts below). Retry in-call from the already-downloaded + // history: the raced ops become durable in the next flush barrier and fold + // into preservedLocalOps via the resume branch, so each retry converges — + // important while e.g. active time tracking dispatches continuously, where + // waiting for the next sync trigger would re-download everything per + // attempt. In-memory reuse is safe: WS downloads and immediate uploads are + // gated while the rebuild marker is set. On exhaustion, the persisted + // marker hands over to the next-sync resume path as before. + const MAX_CAPTURE_RACE_ATTEMPTS = 3; + let isCrashResume = options?.isCrashResume ?? false; + for (let attempt = 1; ; attempt++) { + try { + // 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 (isCrashResume) { + // Keep the first attempt's pre-replace backup (see option JSDoc). + savedAt = (await this.opLogStore.loadImportBackup())?.savedAt; capturedBackupSavedAt = savedAt; - } catch (e) { - OpLog.warn( - 'OperationLogSyncService: Pre-replace safety backup failed; aborting force download.', - { name: (e as Error | undefined)?.name }, + const marker = await this.opLogStore.loadRawRebuildIncomplete(); + const liveLocalOps = (await this.opLogStore.getUnsynced()).map( + (entry) => entry.op, ); - throw new Error( - 'Pre-replace safety backup failed; aborting to preserve local state.', + preservedLocalOps = this._mergeOperationsById( + marker?.preservedLocalOps ?? [], + liveLocalOps, ); + } else { + try { + savedAt = await this.backupService.captureImportBackup(); + capturedBackupSavedAt = savedAt; + } catch (e) { + OpLog.warn( + 'OperationLogSyncService: Pre-replace safety backup failed; aborting force download.', + { name: (e as Error | undefined)?.name }, + ); + throw new Error( + '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, - preservedLocalOps, - }); - 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, + preservedLocalOps, + }); + 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: Force download received snapshotState. Hydrating...', + 'OperationLogSyncService: Replaced local persistence with remote baseline.', ); - // 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', - ); + // 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: Wrote ${appendResult.writtenOps.length} snapshot ops to IndexedDB ` + - 'after force-download hydration.' + - (appendResult.skippedCount > 0 - ? ` Skipped ${appendResult.skippedCount} duplicate(s).` - : ''), + '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).` + : ''), + ); + } + + await this._restorePreservedLocalOps(preservedLocalOps); + + await this._assertNoCaptureRacedWithRebuild(); + + // Update lastServerSeq after hydration + if (result.latestServerSeq !== undefined) { + await syncProvider.setLastServerSeq(result.latestServerSeq); + } + + await this._completeRawRebuild(); + + 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.', ); } @@ -1524,7 +1597,7 @@ export class OperationLogSyncService { await this._assertNoCaptureRacedWithRebuild(); - // Update lastServerSeq after hydration + // Update lastServerSeq if (result.latestServerSeq !== undefined) { await syncProvider.setLastServerSeq(result.latestServerSeq); } @@ -1532,67 +1605,33 @@ export class OperationLogSyncService { await this._completeRawRebuild(); OpLog.normal( - 'OperationLogSyncService: Force download snapshot hydration complete.', + `OperationLogSyncService: Force download complete. Rebuilt from ${migratedRemoteOps.length} ops.`, ); 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.', + }); + break; + } catch (e) { + if ( + e instanceof CaptureRacedRebuildError && + attempt < MAX_CAPTURE_RACE_ATTEMPTS + ) { + OpLog.warn( + `OperationLogSyncService: Local capture raced the rebuild; retrying phase 2 in-call ` + + `(attempt ${attempt}/${MAX_CAPTURE_RACE_ATTEMPTS}).`, ); + // The replacement committed before the assert threw, so the marker is + // set: re-enter through the crash-resume branch, which keeps the first + // attempt's backup and merges the newly-durable raced ops. + isCrashResume = true; + continue; } - - await this._restorePreservedLocalOps(preservedLocalOps); - - await this._assertNoCaptureRacedWithRebuild(); - - // Update lastServerSeq - if (result.latestServerSeq !== undefined) { - await syncProvider.setLastServerSeq(result.latestServerSeq); + // Final failure only — showing this per aborted attempt would churn the + // single snack slot. + if (replacementCommitted && capturedBackupSavedAt !== undefined) { + this._showRestorePreviousDataSnack(capturedBackupSavedAt); } - - await this._completeRawRebuild(); - - OpLog.normal( - `OperationLogSyncService: Force download complete. Rebuilt from ${migratedRemoteOps.length} ops.`, - ); - return savedAt; - }); - } catch (e) { - if (replacementCommitted && capturedBackupSavedAt !== undefined) { - this._showRestorePreviousDataSnack(capturedBackupSavedAt); + throw e; } - throw e; } // On a crash resume without a surviving backup there is nothing to offer. @@ -1624,6 +1663,38 @@ export class OperationLogSyncService { } private async _assertNoIncompleteRemoteOperations(): Promise { + const state = await this._readIncompleteRemoteOperationsState(); + if (!state.isBlocked) { + return; + } + + // One in-session repair attempt when the ONLY blockers are quarantined + // archive failures: their reducers already committed, and the archive-only + // retry is idempotent (ARCHIVE_AFFECTING_ACTION_TYPES invariant). Without + // this, a transient archive failure wedges sync until the next app start — + // the error snack's "restart" advice — even though a retry would succeed + // immediately. Never attempted while a raw rebuild is incomplete or + // reducer-uncommitted `pending` rows exist (retrying those would be wrong). + if (!state.isRawRebuildIncomplete && state.pendingCount === 0) { + await this.injector.get(OperationLogHydratorService).retryFailedRemoteOps(); + const recheck = await this._readIncompleteRemoteOperationsState(); + if (!recheck.isBlocked) { + OpLog.normal( + 'OperationLogSyncService: In-session archive retry cleared the incomplete-remote gate.', + ); + return; + } + } + + throw new IncompleteRemoteOperationsError(); + } + + private async _readIncompleteRemoteOperationsState(): Promise<{ + isBlocked: boolean; + isRawRebuildIncomplete: boolean; + pendingCount: number; + failedCount: number; + }> { const [isRawRebuildIncomplete, pendingRemoteOps, failedRemoteOps] = await Promise.all( [ this.opLogStore.isRawRebuildIncomplete(), @@ -1631,13 +1702,15 @@ export class OperationLogSyncService { this.opLogStore.getFailedRemoteOps(), ], ); - if ( - isRawRebuildIncomplete || - pendingRemoteOps.length > 0 || - failedRemoteOps.length > 0 - ) { - throw new IncompleteRemoteOperationsError(); - } + return { + isBlocked: + isRawRebuildIncomplete || + pendingRemoteOps.length > 0 || + failedRemoteOps.length > 0, + isRawRebuildIncomplete, + pendingCount: pendingRemoteOps.length, + failedCount: failedRemoteOps.length, + }; } private async _resumeInterruptedRawRebuild( @@ -1669,9 +1742,7 @@ export class OperationLogSyncService { private _assertNoCaptureRacedWithRebuild(): void { if (this.writeFlushService.hasPendingWrites()) { - throw new Error( - 'USE_REMOTE incomplete: a local change arrived during the rebuild and will be restored on retry.', - ); + throw new CaptureRacedRebuildError(); } } @@ -1744,13 +1815,19 @@ export class OperationLogSyncService { ); } if (version > CURRENT_SCHEMA_VERSION) { - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.VERSION_TOO_OLD, - actionStr: T.PS.UPDATE_APP, - actionFn: () => - window.open('https://super-productivity.com/download', '_blank'), - }); + if ( + !this._hasWarnedRebuildVersionBlockThisSession && + !this.snackService.hasPendingPersistentAction() + ) { + this._hasWarnedRebuildVersionBlockThisSession = true; + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.VERSION_TOO_OLD, + actionStr: T.PS.UPDATE_APP, + actionFn: () => + window.open('https://super-productivity.com/download', '_blank'), + }); + } throw new Error( 'USE_REMOTE aborted: remote history contains ops from a newer schema version — update the app first.', ); @@ -1801,6 +1878,18 @@ export class OperationLogSyncService { }); } + /** + * Boot-time entry point for the interrupted-rebuild affordance (see + * StartupService): the user woke up on the rebuild baseline, not their data. + * A running sync resumes the rebuild by itself, but when none will run + * (offline, or sync disabled facing the "emptied" app) the pre-replace + * backup would have no visible entry point. Non-destructive — only surfaces + * the persistent restore snack, deduped against a visible recovery action. + */ + async offerInterruptedRebuildRecovery(): Promise { + await this._offerStrandedRebuildBackup(); + } + /** * Offer the pre-replace Undo after an interrupted USE_REMOTE rebuild whose * resume could not finish. The first attempt already committed the destructive diff --git a/src/app/op-log/sync/remote-ops-processing.service.spec.ts b/src/app/op-log/sync/remote-ops-processing.service.spec.ts index d427953960..b72cbb6d00 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.spec.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.spec.ts @@ -75,7 +75,11 @@ describe('RemoteOpsProcessingService', () => { 'getCurrentVersion', 'migrateOperation', ]); - snackServiceSpy = jasmine.createSpyObj('SnackService', ['open']); + snackServiceSpy = jasmine.createSpyObj('SnackService', [ + 'open', + 'hasPendingPersistentAction', + ]); + snackServiceSpy.hasPendingPersistentAction.and.returnValue(false); opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [ 'getUnsynced', 'hasOp', @@ -670,6 +674,20 @@ describe('RemoteOpsProcessingService', () => { }); }); + it('should not replace a visible persistent recovery action with the version-block snack', async () => { + snackServiceSpy.hasPendingPersistentAction.and.returnValue(true); + + await service.processRemoteOps([{ id: 'op1', schemaVersion: 2 } as Operation]); + + expect(snackServiceSpy.open).not.toHaveBeenCalled(); + + // The latch stayed unset, so a later retry (after the recovery action + // resolves) still warns the user. + snackServiceSpy.hasPendingPersistentAction.and.returnValue(false); + await service.processRemoteOps([{ id: 'op2', schemaVersion: 2 } as Operation]); + expect(snackServiceSpy.open).toHaveBeenCalledTimes(1); + }); + it('should show the version-block error only once per session', async () => { // Current version is 1 (set in beforeEach) const remoteOps1: Operation[] = [{ id: 'op1', schemaVersion: 2 } as Operation]; diff --git a/src/app/op-log/sync/remote-ops-processing.service.ts b/src/app/op-log/sync/remote-ops-processing.service.ts index 4d5a92a8d8..95c03e1337 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.ts @@ -396,6 +396,12 @@ export class RemoteOpsProcessingService { | 'INVALID_SCHEMA_VERSION' | 'MIGRATION_FAILED', ): void { + if (this.snackService.hasPendingPersistentAction()) { + // Never replace a visible persistent recovery action (e.g. the USE_REMOTE + // Undo — the only entry point to the pre-replace backup). The block + // persists, so the latch stays unset and a later retry re-warns. + return; + } if (reason === 'MIGRATION_FAILED' || reason === 'INVALID_SCHEMA_VERSION') { if (!this._hasWarnedMigrationFailureThisSession) { this._hasWarnedMigrationFailureThisSession = true; @@ -408,15 +414,22 @@ export class RemoteOpsProcessingService { } if (!this._hasWarnedVersionBlockThisSession) { this._hasWarnedVersionBlockThisSession = true; - this.snackService.open({ - type: 'ERROR', - msg: - reason === 'VERSION_UNSUPPORTED' - ? T.F.SYNC.S.VERSION_UNSUPPORTED - : T.F.SYNC.S.VERSION_TOO_OLD, - actionStr: T.PS.UPDATE_APP, - actionFn: () => window.open('https://super-productivity.com/download', '_blank'), - }); + if (reason === 'VERSION_UNSUPPORTED') { + // Below-minimum data: updating THIS device cannot help, so no + // download action — the fix lives on the device that produced it. + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.VERSION_UNSUPPORTED, + }); + } else { + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.VERSION_TOO_OLD, + actionStr: T.PS.UPDATE_APP, + actionFn: () => + window.open('https://super-productivity.com/download', '_blank'), + }); + } } } diff --git a/src/app/op-log/sync/sync-import-conflict-gate.service.ts b/src/app/op-log/sync/sync-import-conflict-gate.service.ts index 5b2abd6763..60198ffb8f 100644 --- a/src/app/op-log/sync/sync-import-conflict-gate.service.ts +++ b/src/app/op-log/sync/sync-import-conflict-gate.service.ts @@ -93,9 +93,13 @@ export class SyncImportConflictGateService { /** * @param options.preCapturedPendingOps - Exact pending ops selected by the upload - * round. The piggyback path unions this snapshot with a live read so it - * protects both accepted work from that upload and work created while the - * network request was in flight. + * round, unioned with a live read. NOTE: accepted ops of the current round + * are still in the live set when this gate runs (acknowledgement is + * deferred until piggyback processing commits), so the union is NOT what + * protects them. Its own coverage is the narrower set of ops removed from + * the live set mid-upload: a local SYNC_IMPORT deleted on + * SYNC_IMPORT_EXISTS, full-state ops perma-rejected by the server, and a + * concurrent tab marking ops synced between lock release and this read. */ async checkIncomingFullStateConflict( incomingOps: Operation[], diff --git a/src/app/op-log/testing/integration/migration-handling.integration.spec.ts b/src/app/op-log/testing/integration/migration-handling.integration.spec.ts index 09067ad346..aae2c542ab 100644 --- a/src/app/op-log/testing/integration/migration-handling.integration.spec.ts +++ b/src/app/op-log/testing/integration/migration-handling.integration.spec.ts @@ -30,7 +30,11 @@ describe('Migration Handling Integration', () => { let operationApplierSpy: jasmine.SpyObj; beforeEach(async () => { - snackServiceSpy = jasmine.createSpyObj('SnackService', ['open']); + snackServiceSpy = jasmine.createSpyObj('SnackService', [ + 'open', + 'hasPendingPersistentAction', + ]); + snackServiceSpy.hasPendingPersistentAction.and.returnValue(false); operationApplierSpy = jasmine.createSpyObj('OperationApplierService', [ 'applyOperations', ]); diff --git a/src/assets/i18n/en.json b/src/assets/i18n/en.json index 6e39ab4cee..9f8a73b31b 100644 --- a/src/assets/i18n/en.json +++ b/src/assets/i18n/en.json @@ -1586,7 +1586,7 @@ "LOCK_TIMEOUT_ERROR": "Sync timed out waiting for a previous operation to finish. Please try again.", "LWW_CONFLICTS_AUTO_RESOLVED": "Sync conflicts auto-resolved: {{localWins}} local win(s), {{remoteWins}} remote win(s)", "LOCAL_FILE_RESELECT_REQUIRED": "Local file sync needs the sync folder to be selected again after a security update. Open Sync settings and choose the folder once more.", - "MIGRATION_FAILED": "Some sync data could not be migrated and was skipped. This may indicate data corruption or a bug.", + "MIGRATION_FAILED": "Some synced changes could not be migrated. Sync is paused before them to protect your data; updating the app may fix this.", "NETWORK_ERROR": "Sync request failed because of a temporary network problem. Your changes remain local and sync will retry.", "NEWER_VERSION_AVAILABLE": "Some changes are from a newer app version. Consider updating for full compatibility.", "OAUTH_STATE_INVALID": "OAuth state mismatch — the callback URL did not match this device's request. Please try authenticating again.", @@ -1615,7 +1615,7 @@ "UPLOAD_OPS_REJECTED": "{{count}} operation(s) were permanently rejected by the server and won't be synced.", "VECTOR_CLOCK_LIMIT_REACHED": "Sync tidied up old device entries ({{originalSize}} → {{maxSize}}). This is normal — no action needed.", "VERSION_TOO_OLD": "Your app version is too old for the synced data. Please update!", - "VERSION_UNSUPPORTED": "Cannot sync: data requires a newer app version. Please update.", + "VERSION_UNSUPPORTED": "Cannot sync: some synced data is from an app version that is no longer supported. Sync is paused before it; updating the app on your other devices may help.", "WEB_CRYPTO_NOT_AVAILABLE": "Encryption is not available on this device. On Android, encryption requires a secure context (HTTPS) which is not available. Please disable encryption in sync settings or sync from a desktop web browser." }, "SAFETY_BACKUP": {