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 cba00d4ac5..b28d9c05c4 100644 --- a/src/app/op-log/capture/operation-log.effects.spec.ts +++ b/src/app/op-log/capture/operation-log.effects.spec.ts @@ -805,7 +805,7 @@ describe('OperationLogEffects', () => { }); it('should do nothing when no deferred actions are buffered', async () => { - await effects.processDeferredActions(); + await expectAsync(effects.processDeferredActions()).toBeResolved(); expect(mockOpLogStore.appendWithVectorClockUpdate).not.toHaveBeenCalled(); }); @@ -893,12 +893,18 @@ describe('OperationLogEffects', () => { new Error('transient failure'), ); - await effects.processDeferredActions(); + await expectAsync(effects.processDeferredActions()).toBeRejected(); // 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]); + expect(mockSnackService.open).toHaveBeenCalledWith( + jasmine.objectContaining({ + msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED, + actionStr: T.G.DISMISS, + }), + ); mockOpLogStore.appendWithVectorClockUpdate.calls.reset(); mockOpLogStore.appendWithVectorClockUpdate.and.resolveTo(2); @@ -931,7 +937,10 @@ describe('OperationLogEffects', () => { ).toBe(ActionType.TASK_SHARED_UPDATE); expect(getDeferredActions()).toEqual([]); expect(mockSnackService.open).toHaveBeenCalledWith( - jasmine.objectContaining({ msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED }), + jasmine.objectContaining({ + msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED, + actionStr: T.G.DISMISS, + }), ); }); @@ -1022,7 +1031,9 @@ describe('OperationLogEffects', () => { new DOMException('Quota exceeded', 'QuotaExceededError'), ); - await effects.processDeferredActions({ callerHoldsOperationLogLock: true }); + await expectAsync( + effects.processDeferredActions({ callerHoldsOperationLogLock: true }), + ).toBeRejected(); expect(mockCompactionService.emergencyCompact).not.toHaveBeenCalled(); expect(mockLockService.request).not.toHaveBeenCalledWith( @@ -1055,7 +1066,9 @@ describe('OperationLogEffects', () => { new DOMException('Quota exceeded', 'QuotaExceededError'), ); - await effects.processDeferredActions({ callerHoldsOperationLogLock: true }); + await expectAsync( + effects.processDeferredActions({ callerHoldsOperationLogLock: true }), + ).toBeRejected(); // 1. The bail path actually ran (proves handleQuotaExceeded was invoked // AND took the caller-holds-lock branch — not some other code path). diff --git a/src/app/op-log/capture/operation-log.effects.ts b/src/app/op-log/capture/operation-log.effects.ts index 68c4b70165..76edf6a342 100644 --- a/src/app/op-log/capture/operation-log.effects.ts +++ b/src/app/op-log/capture/operation-log.effects.ts @@ -305,6 +305,11 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // (recordCriticalErrorTime) — that is fed by GlobalErrorHandler and the // validateState seam only; snackbar-surfaced conditions are out of scope // by design. See util/critical-error-signal.ts. + if (isDeferredWrite) { + throw new PermanentDeferredWriteError( + `Deferred action ${action.type} has an invalid payload.`, + ); + } this.snackService.open({ type: 'ERROR', msg: T.F.SYNC.S.INVALID_OPERATION_PAYLOAD, @@ -313,11 +318,6 @@ export class OperationLogEffects implements DeferredLocalActionsPort { window.location.reload(); }, }); - if (isDeferredWrite) { - throw new PermanentDeferredWriteError( - `Deferred action ${action.type} has an invalid payload.`, - ); - } return; // Skip persisting invalid operation } @@ -409,7 +409,9 @@ export class OperationLogEffects implements DeferredLocalActionsPort { // #8306: the effect path catches this throw in writeOperationFromEffect // — the shared persistOperation$ stream must NOT be torn down by one // failed write — while still showing the sticky snackbar below. - this.notifyUserAndTriggerRollback(); + if (!isDeferredWrite) { + this.notifyUserAndTriggerRollback(); + } throw e; } if (this.isQuotaExceededError(e)) { @@ -418,13 +420,17 @@ export class OperationLogEffects implements DeferredLocalActionsPort { OpLog.err( 'OperationLogEffects: Quota exceeded during retry - aborting to prevent loop', ); - this.notifyUserAndTriggerRollback(); + if (!isDeferredWrite) { + this.notifyUserAndTriggerRollback(); + } } else { await this.handleQuotaExceeded(action, isDeferredWrite, options); return; } } else { - this.notifyUserAndTriggerRollback(); + if (!isDeferredWrite) { + this.notifyUserAndTriggerRollback(); + } } throw e; } @@ -687,6 +693,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { const MAX_RETRIES = 3; const BASE_DELAY_MS = 100; let failedCount = 0; + let transientFailure: unknown; for (const action of deferredActions) { let lastError: unknown; @@ -728,8 +735,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { 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. + // and re-raise the failure snack on every future sync window. acknowledgeDeferredAction(action); OpLog.err('OperationLogEffects: Abandoning permanently invalid deferred action', { actionType: action.type, @@ -749,6 +755,7 @@ export class OperationLogEffects implements DeferredLocalActionsPort { errorName: (lastError as Error | undefined)?.name, }, ); + transientFailure = lastError; break; } @@ -757,14 +764,18 @@ export class OperationLogEffects implements DeferredLocalActionsPort { this.snackService.open({ type: 'ERROR', msg: T.F.SYNC.S.DEFERRED_ACTION_FAILED, - actionStr: T.PS.RELOAD, - actionFn: (): void => { - window.location.reload(); - }, + actionStr: T.G.DISMISS, config: { duration: 0, // Sticky - don't auto-dismiss critical errors }, }); } + + if (transientFailure !== undefined) { + throw new Error( + 'Deferred local actions remain unpersisted after retry exhaustion.', + { cause: transientFailure }, + ); + } } } diff --git a/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts b/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts index 1fe578bbba..e7d0dd52c9 100644 --- a/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts +++ b/src/app/op-log/sync/operation-log-lock-reentry.regression.spec.ts @@ -218,14 +218,16 @@ describe('regression #7700: operation-log lock reentry', () => { // Hold the lock comfortably longer than the full retry budget: // 3 attempts × ~1000ms timeout + 100ms + 200ms backoffs ≈ 3300ms. // 10000ms is generous. - await lockService.request( - LOCK_NAMES.OPERATION_LOG, - async () => { - // Caller forgot the flag — same as a buggy refactor. - await effects.processDeferredActions(); - }, - 10000, - ); + await expectAsync( + lockService.request( + LOCK_NAMES.OPERATION_LOG, + async () => { + // Caller forgot the flag — same as a buggy refactor. + await effects.processDeferredActions(); + }, + 10000, + ), + ).toBeRejected(); // Action was NOT written — no silent persistence. expect(opLogStoreSpy.appendWithVectorClockUpdate).not.toHaveBeenCalled(); 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 1189c3f3a4..ef3a282ebe 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 @@ -6,6 +6,7 @@ import { SnackService } from '../../core/snack/snack.service'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { VectorClockService } from './vector-clock.service'; import { OperationApplierService } from '../apply/operation-applier.service'; +import { OperationLogEffects } from '../capture/operation-log.effects'; import { ConflictResolutionService } from './conflict-resolution.service'; import { ValidateStateService } from '../validation/validate-state.service'; import { SyncSessionValidationService } from './sync-session-validation.service'; @@ -58,6 +59,7 @@ describe('OperationLogSyncService', () => { let validateStateServiceSpy: jasmine.SpyObj; let lockServiceSpy: jasmine.SpyObj; let operationApplierSpy: jasmine.SpyObj; + let operationLogEffectsSpy: jasmine.SpyObj; beforeEach(() => { snackServiceSpy = jasmine.createSpyObj('SnackService', [ @@ -67,6 +69,8 @@ describe('OperationLogSyncService', () => { snackServiceSpy.hasPendingPersistentAction.and.returnValue(false); opLogStoreSpy = jasmine.createSpyObj('OperationLogStoreService', [ 'getUnsynced', + 'getPendingRemoteOps', + 'getFailedRemoteOps', 'loadStateCache', 'getLastSeq', 'getOpById', @@ -85,6 +89,8 @@ describe('OperationLogSyncService', () => { ]); opLogStoreSpy.hasSyncedOps.and.resolveTo(true); opLogStoreSpy.getUnsynced.and.resolveTo([]); + opLogStoreSpy.getPendingRemoteOps.and.resolveTo([]); + opLogStoreSpy.getFailedRemoteOps.and.resolveTo([]); opLogStoreSpy.markSynced.and.resolveTo(); opLogStoreSpy.setVectorClock.and.resolveTo(); opLogStoreSpy.clearFullStateOps.and.resolveTo(); @@ -188,6 +194,10 @@ describe('OperationLogSyncService', () => { 'applyOperations', ]); operationApplierSpy.applyOperations.and.resolveTo({ appliedOps: [] }); + operationLogEffectsSpy = jasmine.createSpyObj('OperationLogEffects', [ + 'processDeferredActions', + ]); + operationLogEffectsSpy.processDeferredActions.and.resolveTo(); TestBed.configureTestingModule({ providers: [ @@ -209,6 +219,7 @@ describe('OperationLogSyncService', () => { provide: OperationApplierService, useValue: operationApplierSpy, }, + { provide: OperationLogEffects, useValue: operationLogEffectsSpy }, { provide: ConflictResolutionService, useValue: jasmine.createSpyObj('ConflictResolutionService', [ @@ -315,6 +326,41 @@ describe('OperationLogSyncService', () => { }); describe('uploadPendingOps', () => { + it('should drain deferred local actions before selecting pending uploads', async () => { + const callOrder: string[] = []; + writeFlushServiceSpy.flushPendingWrites.and.callFake(async () => { + callOrder.push('flush'); + }); + operationLogEffectsSpy.processDeferredActions.and.callFake(async () => { + callOrder.push('deferred'); + }); + uploadServiceSpy.uploadPendingOps.and.callFake(async () => { + callOrder.push('upload'); + return { + uploadedCount: 0, + piggybackedOps: [], + rejectedCount: 0, + rejectedOps: [], + }; + }); + + await service.uploadPendingOps({} as OperationSyncCapable); + + expect(callOrder).toEqual(['flush', 'deferred', 'flush', 'upload']); + }); + + it('should block upload while a remote operation is incompletely applied', async () => { + opLogStoreSpy.getPendingRemoteOps.and.resolveTo([ + { applicationStatus: 'pending' } as OperationLogEntry, + ]); + + await expectAsync( + service.uploadPendingOps({} as OperationSyncCapable), + ).toBeRejected(); + + expect(uploadServiceSpy.uploadPendingOps).not.toHaveBeenCalled(); + }); + it('should return localWinOpsCreated: 0 when no piggybacked ops', async () => { opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([])); uploadServiceSpy.uploadPendingOps.and.returnValue( @@ -908,6 +954,18 @@ describe('OperationLogSyncService', () => { }); describe('downloadRemoteOps', () => { + it('should block download while a prior remote operation is incompletely applied', async () => { + opLogStoreSpy.getFailedRemoteOps.and.resolveTo([ + { applicationStatus: 'archive_pending' } as OperationLogEntry, + ]); + + await expectAsync( + service.downloadRemoteOps({} as OperationSyncCapable), + ).toBeRejected(); + + expect(downloadServiceSpy.downloadRemoteOps).not.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. @@ -3962,7 +4020,7 @@ describe('OperationLogSyncService', () => { const result = await service.downloadRemoteOps(mockProvider); - expect(events.slice(0, 2)).toEqual(['flush', 'getUnsynced']); + expect(events.slice(0, 4)).toEqual(['flush', 'flush', 'flush', 'getUnsynced']); expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ incomingSyncImport, ]); @@ -4296,7 +4354,7 @@ describe('OperationLogSyncService', () => { const result = await service.uploadPendingOps(mockProvider); - expect(writeFlushServiceSpy.flushPendingWrites).toHaveBeenCalledTimes(2); + expect(writeFlushServiceSpy.flushPendingWrites).toHaveBeenCalledTimes(3); expect(opLogStoreSpy.getUnsynced).toHaveBeenCalled(); expect(result.kind).toBe('completed'); }); 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 0aff4ddf64..f6d453f425 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -1,4 +1,4 @@ -import { inject, Injectable } from '@angular/core'; +import { inject, Injectable, Injector } from '@angular/core'; import { Store } from '@ngrx/store'; import { planSnapshotHydration } from '@sp/sync-core'; import { @@ -61,6 +61,7 @@ import { } from '../../features/config/local-only-sync-settings.util'; import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const'; import { OperationApplierService } from '../apply/operation-applier.service'; +import { processDeferredActions } from './process-deferred-actions-flush.util'; /** * Orchestrates synchronization of the Operation Log with remote storage. @@ -151,6 +152,7 @@ export class OperationLogSyncService { private syncImportConflictCoordinator = inject(SyncImportConflictCoordinatorService); private providerManager = inject(SyncProviderManager); private operationApplier = inject(OperationApplierService); + private injector = inject(Injector); /** * Checks if this client is "wholly fresh" - meaning it has never synced before @@ -206,7 +208,8 @@ export class OperationLogSyncService { // The effect that writes operations uses concatMap for sequential processing, // but if sync is triggered before all operations are written to IndexedDB, // we would upload an incomplete set. This flush waits for all queued writes. - await this.writeFlushService.flushPendingWrites(); + await this._flushLocalWritesIncludingDeferredActions(); + await this._assertNoIncompleteRemoteOperations(); // Capture never-synced status before the upload runs. The orchestrator passes a // value captured even earlier (pre-download, since download persists synced ops); @@ -470,6 +473,9 @@ export class OperationLogSyncService { return { kind: 'snapshot_hydrated' }; } + await this._flushLocalWritesIncludingDeferredActions(); + await this._assertNoIncompleteRemoteOperations(); + const result = await this.downloadService.downloadRemoteOps(syncProvider, options); // FIX #6571: Check download success before processing results. @@ -1524,6 +1530,26 @@ export class OperationLogSyncService { return merged; } + private async _flushLocalWritesIncludingDeferredActions(): Promise { + await this.writeFlushService.flushPendingWrites(); + await processDeferredActions(this.injector, false); + // Deferred writes are awaited directly, but this second barrier also + // catches ordinary actions dispatched while their drain was running. + await this.writeFlushService.flushPendingWrites(); + } + + private async _assertNoIncompleteRemoteOperations(): Promise { + const [pendingRemoteOps, failedRemoteOps] = await Promise.all([ + this.opLogStore.getPendingRemoteOps(), + this.opLogStore.getFailedRemoteOps(), + ]); + if (pendingRemoteOps.length > 0 || failedRemoteOps.length > 0) { + throw new Error( + 'Sync blocked because previously downloaded operations are not fully applied. Restart the app to retry recovery.', + ); + } + } + /** * Replays edits captured after an interrupted rebuild on top of the complete * authoritative history. They stay local/unsynced so the next upload carries diff --git a/src/app/op-log/sync/process-deferred-actions-flush.util.ts b/src/app/op-log/sync/process-deferred-actions-flush.util.ts index 326b8cba8c..d1b763d5f9 100644 --- a/src/app/op-log/sync/process-deferred-actions-flush.util.ts +++ b/src/app/op-log/sync/process-deferred-actions-flush.util.ts @@ -19,7 +19,7 @@ import { OperationLogEffects } from '../capture/operation-log.effects'; * the sp_op_log lock; the flush then runs inline rather * than re-acquiring (non-reentrant lock would deadlock). */ -export const processDeferredActionsAfterRemoteApply = async ( +export const processDeferredActions = async ( injector: Injector, callerHoldsOperationLogLock: boolean, ): Promise => { @@ -27,3 +27,5 @@ export const processDeferredActionsAfterRemoteApply = async ( callerHoldsOperationLogLock, }); }; + +export const processDeferredActionsAfterRemoteApply = processDeferredActions; diff --git a/src/assets/i18n/en.json b/src/assets/i18n/en.json index c51dd49868..6ced86a079 100644 --- a/src/assets/i18n/en.json +++ b/src/assets/i18n/en.json @@ -1549,7 +1549,7 @@ "CONFLICT_RESOLUTION_FAILED": "Sync conflict resolution failed. Please reload.", "DATA_REPAIRED": "Sync cleanup: {{count}} reference(s) resolved", "DECRYPTION_FAILED": "Failed to decrypt synced data. Please check your encryption password.", - "DEFERRED_ACTION_FAILED": "Some changes made during sync could not be saved. Please reload to recover.", + "DEFERRED_ACTION_FAILED": "Some changes made during sync could not be saved yet. Sync will retry automatically; keep the app open.", "DISABLE_ENCRYPTION_FAILED": "Failed to disable encryption: {{message}}", "ENABLE_ENCRYPTION_FAILED": "Failed to enable encryption: {{message}}", "ENCRYPTION_DISABLED_ON_OTHER_DEVICE": "Encryption was disabled on another device. Your sync settings have been updated.",