diff --git a/e2e/tests/sync/supersync-multi-tab.spec.ts b/e2e/tests/sync/supersync-multi-tab.spec.ts index d29ef159de..f973651aaf 100644 --- a/e2e/tests/sync/supersync-multi-tab.spec.ts +++ b/e2e/tests/sync/supersync-multi-tab.spec.ts @@ -10,6 +10,142 @@ import { import { SuperSyncPage } from '../../pages/supersync.page'; import { WorkViewPage } from '../../pages/work-view.page'; import { waitForAppReady } from '../../utils/waits'; +import { ImportPage } from '../../pages/import.page'; +import type { BrowserContext, Page } from '@playwright/test'; + +const UPLOAD_LOCK_NAME = 'sp_op_log_upload'; + +const installSecondTabTestInit = async (context: BrowserContext): Promise => { + await context.addInitScript(() => { + localStorage.setItem('SUP_ONBOARDING_PRESET_DONE', 'true'); + localStorage.setItem('SUP_ONBOARDING_HINTS_DONE', 'true'); + localStorage.setItem('SUP_IS_SHOW_TOUR', 'true'); + localStorage.setItem('SUP_EXAMPLE_TASKS_CREATED', 'true'); + + const testGlobal = globalThis as typeof globalThis & { + __SP_E2E_BLOCK_AUTO_SYNC?: boolean; + __SP_E2E_BLOCK_IMMEDIATE_UPLOAD?: boolean; + __SP_E2E_BLOCK_WS_DOWNLOAD?: boolean; + }; + testGlobal.__SP_E2E_BLOCK_AUTO_SYNC = true; + testGlobal.__SP_E2E_BLOCK_IMMEDIATE_UPLOAD = true; + testGlobal.__SP_E2E_BLOCK_WS_DOWNLOAD = true; + + const NativeBroadcastChannel = globalThis.BroadcastChannel; + class MultiTabTestBroadcastChannel extends NativeBroadcastChannel { + private readonly _isSingleInstanceChannel: boolean; + + constructor(name: string) { + super(name); + this._isSingleInstanceChannel = name === 'superProductivityTab'; + } + + override postMessage(message: unknown): void { + if (this._isSingleInstanceChannel) { + return; + } + super.postMessage(message); + } + } + + Object.defineProperty(globalThis, 'BroadcastChannel', { + configurable: true, + writable: true, + value: MultiTabTestBroadcastChannel, + }); + }); +}; + +const acquireUploadLock = async (page: Page): Promise => { + await page.evaluate((lockName) => { + const testGlobal = globalThis as typeof globalThis & { + __SP_E2E_RELEASE_UPLOAD_LOCK?: () => void; + __SP_E2E_UPLOAD_LOCK_HELD?: boolean; + __SP_E2E_UPLOAD_LOCK_ERROR?: string; + }; + let releaseLock: () => void = () => undefined; + const releasePromise = new Promise((resolve) => { + releaseLock = resolve; + }); + testGlobal.__SP_E2E_RELEASE_UPLOAD_LOCK = releaseLock; + testGlobal.__SP_E2E_UPLOAD_LOCK_HELD = false; + void navigator.locks + .request(lockName, async () => { + testGlobal.__SP_E2E_UPLOAD_LOCK_HELD = true; + try { + await releasePromise; + } finally { + testGlobal.__SP_E2E_UPLOAD_LOCK_HELD = false; + } + }) + .catch((error: unknown) => { + testGlobal.__SP_E2E_UPLOAD_LOCK_ERROR = String(error); + }); + }, UPLOAD_LOCK_NAME); + + await page.waitForFunction(() => { + const testGlobal = globalThis as typeof globalThis & { + __SP_E2E_UPLOAD_LOCK_HELD?: boolean; + __SP_E2E_UPLOAD_LOCK_ERROR?: string; + }; + if (testGlobal.__SP_E2E_UPLOAD_LOCK_ERROR) { + throw new Error(testGlobal.__SP_E2E_UPLOAD_LOCK_ERROR); + } + return testGlobal.__SP_E2E_UPLOAD_LOCK_HELD === true; + }); +}; + +const releaseUploadLock = async (page: Page): Promise => { + await page.evaluate(() => { + const testGlobal = globalThis as typeof globalThis & { + __SP_E2E_RELEASE_UPLOAD_LOCK?: () => void; + }; + testGlobal.__SP_E2E_RELEASE_UPLOAD_LOCK?.(); + }); +}; + +const waitForQueuedUploadLock = async (page: Page): Promise => { + await page.waitForFunction( + async (lockName) => { + const snapshot = await navigator.locks.query(); + return snapshot.pending?.some((lock) => lock.name === lockName) ?? false; + }, + UPLOAD_LOCK_NAME, + { timeout: 20000 }, + ); +}; + +const getUnsyncedOperationCount = async (page: Page): Promise => + page.evaluate(async () => { + const db = await new Promise((resolve, reject) => { + const request = indexedDB.open('SUP_OPS'); + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); + + try { + const entries = await new Promise<{ syncedAt?: number; rejectedAt?: number }[]>( + (resolve, reject) => { + const transaction = db.transaction('ops', 'readonly'); + const request = transaction.objectStore('ops').getAll(); + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + transaction.onabort = () => reject(transaction.error); + }, + ); + return entries.filter((entry) => !entry.syncedAt && !entry.rejectedAt).length; + } finally { + db.close(); + } + }); + +const getSuperSyncCursor = async (page: Page): Promise => + page.evaluate(() => { + const key = Object.keys(localStorage).find((candidate) => + candidate.startsWith('super_sync_last_server_seq_'), + ); + return key ? localStorage.getItem(key) : null; + }); /** * SuperSync Multi-Tab Same Account E2E Tests @@ -18,9 +154,8 @@ import { waitForAppReady } from '../../utils/waits'; * don't corrupt data when using SuperSync. * * NOTE: The app prevents two tabs from running simultaneously via BroadcastChannel. - * This test works around that by navigating the inactive tab to about:blank before - * opening the next tab, then reloading it later. This simulates the real-world - * scenario of closing and reopening a tab. + * The convergence test navigates the inactive tab to about:blank. The race test + * suppresses only that startup channel so both tabs can exercise the shared locks. * * Prerequisites: * - super-sync-server running on localhost:1901 with TEST_MODE=true @@ -171,4 +306,83 @@ test.describe('@supersync Multi-Tab Same Account', () => { } } }); + + test('late sibling-tab work blocks a downloaded full-state apply atomically', async ({ + browser, + baseURL, + testRunId, + }) => { + test.setTimeout(180000); + const appUrl = baseURL || 'http://localhost:4242'; + let clientA: SimulatedE2EClient | null = null; + let clientB: SimulatedE2EClient | null = null; + + try { + const user = await createTestUser(testRunId); + const syncConfig = getSuperSyncConfig(user); + + clientA = await createSimulatedClient(browser, appUrl, 'A', testRunId); + await clientA.sync.setupSuperSync(syncConfig); + const baselineTask = `Full-State-Race-Baseline-${testRunId}`; + await clientA.workView.addTask(baselineTask); + await clientA.sync.syncAndWait(); + + clientB = await createSimulatedClient(browser, appUrl, 'B', testRunId); + const tab1 = clientB.page; + const tab1Sync = clientB.sync; + await tab1Sync.setupSuperSync(syncConfig); + await waitForTask(tab1, baselineTask); + + await installSecondTabTestInit(clientB.context); + const tab2 = await clientB.context.newPage(); + await tab2.goto('/'); + await waitForAppReady(tab2); + const tab2WorkView = new WorkViewPage(tab2, `Sibling-${testRunId}`); + await tab2WorkView.waitForTaskList(); + await waitForTask(tab2, baselineTask); + + const importPage = new ImportPage(clientA.page); + await importPage.navigateToImportPage(); + await importPage.importBackupFile(ImportPage.getFixturePath('test-backup.json')); + await clientA.page.goto(clientA.page.url(), { + waitUntil: 'domcontentloaded', + timeout: 30000, + }); + await clientA.page.waitForLoadState('networkidle'); + await clientA.sync.syncAndWait({ useLocal: true }); + + const cursorBeforeDownload = await getSuperSyncCursor(tab1); + expect(cursorBeforeDownload).not.toBeNull(); + const pendingBeforeRace = await getUnsyncedOperationCount(tab2); + expect(pendingBeforeRace).toBe(0); + + await acquireUploadLock(tab1); + await tab1Sync.syncBtn.click(); + await waitForQueuedUploadLock(tab1); + + const lateTask = `Sibling-Late-Task-${testRunId}`; + await tab2WorkView.addTask(lateTask); + await waitForTask(tab2, lateTask); + await expect + .poll(() => getUnsyncedOperationCount(tab2), { timeout: 10000 }) + .toBeGreaterThan(pendingBeforeRace); + + await releaseUploadLock(tab1); + + const conflictDialog = tab1.locator('dialog-sync-import-conflict'); + await expect(conflictDialog).toBeVisible({ timeout: 20000 }); + await conflictDialog.getByRole('button', { name: /cancel/i }).click(); + await expect(conflictDialog).not.toBeVisible({ timeout: 5000 }); + + expect(await getSuperSyncCursor(tab1)).toBe(cursorBeforeDownload); + expect(await getUnsyncedOperationCount(tab2)).toBeGreaterThan(pendingBeforeRace); + await expect(tab2.locator('task', { hasText: lateTask })).toBeVisible(); + } finally { + if (clientB && !clientB.page.isClosed()) { + await releaseUploadLock(clientB.page).catch(() => undefined); + } + if (clientA) await closeClient(clientA); + if (clientB) await closeClient(clientB); + } + }); }); diff --git a/src/app/op-log/apply/hydration-state.service.spec.ts b/src/app/op-log/apply/hydration-state.service.spec.ts index 51d5240012..9b984292f1 100644 --- a/src/app/op-log/apply/hydration-state.service.spec.ts +++ b/src/app/op-log/apply/hydration-state.service.spec.ts @@ -69,6 +69,29 @@ describe('HydrationStateService', () => { }); }); + describe('scoped apply hold', () => { + it('should stay active until every idempotent hold is released', () => { + const releaseOuterHold = service.acquireApplyingRemoteOpsHold(); + const releaseInnerHold = service.acquireApplyingRemoteOpsHold(); + service.startApplyingRemoteOps(); + service.endApplyingRemoteOps(); + + expect(service.isApplyingRemoteOps()).toBeTrue(); + expect(getIsApplyingRemoteOps()).toBeTrue(); + + releaseOuterHold(); + releaseOuterHold(); + + expect(service.isApplyingRemoteOps()).toBeTrue(); + expect(getIsApplyingRemoteOps()).toBeTrue(); + + releaseInnerHold(); + + expect(service.isApplyingRemoteOps()).toBeFalse(); + expect(getIsApplyingRemoteOps()).toBeFalse(); + }); + }); + describe('state transitions', () => { it('should correctly track multiple start/end cycles', () => { expect(service.isApplyingRemoteOps()).toBeFalse(); diff --git a/src/app/op-log/apply/hydration-state.service.ts b/src/app/op-log/apply/hydration-state.service.ts index 69ed496da1..ea380a7b32 100644 --- a/src/app/op-log/apply/hydration-state.service.ts +++ b/src/app/op-log/apply/hydration-state.service.ts @@ -63,6 +63,8 @@ const SYNC_WINDOW_FAILSAFE_MS = 2000; @Injectable({ providedIn: 'root' }) export class HydrationStateService implements RemoteApplyWindowPort { private _isApplyingRemoteOps = signal(false); + private _isDirectApplyActive = false; + private _applyingRemoteOpsHoldCount = 0; private _isInPostSyncCooldown = signal(false); private _isSyncWindowOpen = signal(false); private _cooldownTimer: ReturnType | null = null; @@ -112,8 +114,8 @@ export class HydrationStateService implements RemoteApplyWindowPort { * during this time to prevent superseded vector clocks. */ startApplyingRemoteOps(): void { - this._isApplyingRemoteOps.set(true); - setIsApplyingRemoteOps(true); + this._isDirectApplyActive = true; + this._updateApplyingRemoteOpsState(); } /** @@ -123,8 +125,38 @@ export class HydrationStateService implements RemoteApplyWindowPort { * Re-enables operation capturing for local operations. */ endApplyingRemoteOps(): void { - this._isApplyingRemoteOps.set(false); - setIsApplyingRemoteOps(false); + this._isDirectApplyActive = false; + this._updateApplyingRemoteOpsState(); + } + + /** + * Keeps local persistent actions in the deferred buffer across a wider + * critical section that contains a normal replay apply window. Unlike nested + * start/end calls, a hold survives the replay coordinator's matching end call. + * The returned release function is idempotent. + */ + acquireApplyingRemoteOpsHold(): () => void { + this._applyingRemoteOpsHoldCount++; + this._updateApplyingRemoteOpsState(); + let isReleased = false; + + return (): void => { + if (isReleased) { + return; + } + isReleased = true; + this._applyingRemoteOpsHoldCount = Math.max( + 0, + this._applyingRemoteOpsHoldCount - 1, + ); + this._updateApplyingRemoteOpsState(); + }; + } + + private _updateApplyingRemoteOpsState(): void { + const isApplying = this._isDirectApplyActive || this._applyingRemoteOpsHoldCount > 0; + this._isApplyingRemoteOps.set(isApplying); + setIsApplyingRemoteOps(isApplying); } /** 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 7747c7d827..6235668fa4 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 @@ -1687,9 +1687,12 @@ describe('OperationLogSyncService', () => { // Should NOT throw - incremental sync should process ops normally await expectAsync(service.downloadRemoteOps(mockProvider)).toBeResolved(); - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - remoteOp, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [remoteOp], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); }); it('should protect unsynced user config from file-snapshot replacement', async () => { @@ -4364,9 +4367,12 @@ describe('OperationLogSyncService', () => { expect( syncImportConflictDialogServiceSpy.showConflictDialog, ).not.toHaveBeenCalled(); - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - incomingSyncImport, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [incomingSyncImport], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(mockProvider.setLastServerSeq).toHaveBeenCalledWith(42); expect(result.kind).toBe('ops_processed'); }); @@ -4578,12 +4584,76 @@ describe('OperationLogSyncService', () => { const result = await service.downloadRemoteOps(mockProvider); expect(events.slice(0, 4)).toEqual(['flush', 'flush', 'flush', 'getUnsynced']); - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - incomingSyncImport, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [incomingSyncImport], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(result.kind).toBe('ops_processed'); }); + it('should abort downloaded SYNC_IMPORT apply when meaningful work appears after the initial gate', async () => { + const incomingSyncImport = createIncomingSyncImport(); + const latePendingEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'late-download-local-op', + clientId: 'client-A', + actionType: 'test' as ActionType, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + payload: { title: 'Late local title' }, + vectorClock: { clientA: 2 }, + timestamp: Date.now(), + schemaVersion: 1, + }, + appliedAt: Date.now(), + source: 'local', + }; + + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [incomingSyncImport], + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 42, + }); + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([latePendingEntry]), + ); + remoteOpsProcessingServiceSpy.processRemoteOps.and.callFake( + async (ops, options) => { + const shouldApply = options?.beforeFullStateApply + ? await options.beforeFullStateApply(ops) + : true; + return { + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, + fullStateApplyBlockedByLocalConflict: !shouldApply, + }; + }, + ); + syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL'); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + } as unknown as OperationSyncCapable; + + const result = await service.downloadRemoteOps(mockProvider); + + expect(syncImportConflictDialogServiceSpy.showConflictDialog).toHaveBeenCalledWith( + jasmine.objectContaining({ scenario: 'INCOMING_IMPORT' }), + ); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); + expect(result.kind).toBe('cancelled'); + }); + it('should prompt before replacing pending user config with an incoming SYNC_IMPORT', async () => { const incomingSyncImport = createIncomingSyncImport(); @@ -4740,6 +4810,7 @@ describe('OperationLogSyncService', () => { rejectedCount: 0, rejectedOps: [], pendingAcknowledgementSeqs: [1], + lastServerSeqToPersist: 42, }); // Client has pending ops @@ -4766,6 +4837,7 @@ describe('OperationLogSyncService', () => { const mockProvider = { isReady: () => Promise.resolve(true), + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), } as unknown as OperationSyncCapable; const result = await service.uploadPendingOps(mockProvider); @@ -4777,6 +4849,7 @@ describe('OperationLogSyncService', () => { }), ); expect(opLogStoreSpy.markSynced).not.toHaveBeenCalled(); + expect(mockProvider.setLastServerSeq).not.toHaveBeenCalled(); expect(result.kind).toBe('cancelled'); }); @@ -4878,9 +4951,12 @@ describe('OperationLogSyncService', () => { expect( syncImportConflictDialogServiceSpy.showConflictDialog, ).not.toHaveBeenCalled(); - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - piggybackedSyncImport, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [piggybackedSyncImport], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(result.kind).toBe('completed'); }); @@ -4929,9 +5005,12 @@ describe('OperationLogSyncService', () => { expect(opLogStoreSpy.markRejected).toHaveBeenCalledOnceWith([ 'sync-provider-setup', ]); - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - piggybackedSyncImport, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [piggybackedSyncImport], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(result.kind).toBe('completed'); }); @@ -5023,6 +5102,84 @@ describe('OperationLogSyncService', () => { expect(result.kind).toBe('completed'); }); + it('should abort piggybacked SYNC_IMPORT apply when meaningful work appears after the initial gate', async () => { + const piggybackedSyncImport: Operation = { + id: 'import-1', + clientId: 'client-B', + actionType: ActionType.LOAD_ALL_DATA, + opType: OpType.SyncImport, + entityType: 'ALL', + payload: {}, + vectorClock: { clientB: 5 }, + timestamp: Date.now(), + schemaVersion: 1, + }; + const latePendingEntry: OperationLogEntry = { + seq: 2, + op: { + id: 'late-local-op', + clientId: 'client-A', + actionType: 'test' as ActionType, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + payload: { title: 'Late local title' }, + vectorClock: { clientA: 2 }, + timestamp: Date.now(), + schemaVersion: 1, + }, + appliedAt: Date.now(), + source: 'local', + }; + + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 1, + piggybackedOps: [piggybackedSyncImport], + rejectedCount: 0, + rejectedOps: [], + pendingAcknowledgementSeqs: [1], + }); + // The first read is the initial post-upload gate. The second read is the + // final in-lock recheck immediately before full-state application. + opLogStoreSpy.getUnsynced.and.returnValues( + Promise.resolve([]), + Promise.resolve([latePendingEntry]), + ); + remoteOpsProcessingServiceSpy.processRemoteOps.and.callFake( + async (ops, options) => { + const finalGuard = ( + options as + | { + beforeFullStateApply?: (fullStateOps: Operation[]) => Promise; + } + | undefined + )?.beforeFullStateApply; + const shouldApply = finalGuard ? await finalGuard(ops) : true; + return { + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, + fullStateApplyBlockedByLocalConflict: !shouldApply, + }; + }, + ); + syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL'); + + const mockProvider = { + isReady: () => Promise.resolve(true), + } as unknown as OperationSyncCapable; + + const result = await service.uploadPendingOps(mockProvider); + + expect(syncImportConflictDialogServiceSpy.showConflictDialog).toHaveBeenCalledWith( + jasmine.objectContaining({ scenario: 'INCOMING_IMPORT' }), + ); + expect(opLogStoreSpy.markSynced).not.toHaveBeenCalled(); + expect(result.kind).toBe('cancelled'); + }); + it('should process piggybacked SYNC_IMPORT silently when no meaningful local data', async () => { const piggybackedSyncImport: Operation = { id: 'import-1', @@ -5063,9 +5220,12 @@ describe('OperationLogSyncService', () => { syncImportConflictDialogServiceSpy.showConflictDialog, ).not.toHaveBeenCalled(); // Should process normally via processRemoteOps - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - piggybackedSyncImport, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [piggybackedSyncImport], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(result.kind).not.toBe('cancelled'); }); @@ -5118,9 +5278,12 @@ describe('OperationLogSyncService', () => { syncImportConflictDialogServiceSpy.showConflictDialog, ).not.toHaveBeenCalled(); // Should process normally - expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith([ - piggybackedOp, - ]); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).toHaveBeenCalledWith( + [piggybackedOp], + jasmine.objectContaining({ + beforeFullStateApply: jasmine.any(Function), + }), + ); expect(opLogStoreSpy.markSynced).toHaveBeenCalledOnceWith([1]); expect(events).toEqual(['processRemoteOps', 'markSynced']); expect(result.kind).not.toBe('cancelled'); 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 1d7d8a1ab5..73e639ea07 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -45,7 +45,10 @@ import { SyncImportConflictData, SyncImportConflictResolution, } from './dialog-sync-import-conflict/dialog-sync-import-conflict.component'; -import { SyncImportConflictGateService } from './sync-import-conflict-gate.service'; +import { + IncomingFullStateConflictGateResult, + SyncImportConflictGateService, +} from './sync-import-conflict-gate.service'; import { CURRENT_SCHEMA_VERSION, MIN_SUPPORTED_SCHEMA_VERSION, @@ -59,7 +62,7 @@ import { loadAllData } from '../../root-store/meta/load-all-data.action'; 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 { Operation, OperationLogEntry } from '../core/operation.types'; import { ValidateStateService } from '../validation/validate-state.service'; import { extractEntityKeysFromState } from '../persistence/extract-entity-keys'; import { firstValueFrom } from 'rxjs'; @@ -77,6 +80,10 @@ type RemoteOpsProcessingResult = Awaited< ReturnType >; +type GuardedRemoteOpsProcessingResult = RemoteOpsProcessingResult & { + preApplyFullStateConflict?: IncomingFullStateConflictGateResult; +}; + /** * Orchestrates synchronization of the Operation Log with remote storage. * @@ -358,10 +365,40 @@ export class OperationLogSyncService { result.piggybackedOps, startupCleanupFullStateOpId, startupOpIdsToDiscard, + { + isNeverSynced: isNeverSyncedAtSyncStart, + preCapturedPendingOps: result.selectedPendingOps ?? [], + }, ); localWinOpsCreated = processResult.localWinOpsCreated; // Validation failure (if any) is on the session-validation latch. + if (processResult.preApplyFullStateConflict?.dialogData) { + const { fullStateOp, pendingOps, dialogData } = + processResult.preApplyFullStateConflict; + OpLog.warn( + `OperationLogSyncService: ${fullStateOp?.opType ?? 'Full-state op'} gained ` + + `${pendingOps.length} pending local op(s) before piggyback apply. Showing conflict dialog.`, + ); + const conflictResult = await this._handleSyncImportConflict( + syncProvider, + dialogData, + 'OperationLogSyncService (piggybacked full-state pre-apply recheck)', + ); + if (conflictResult === 'CANCEL') { + return { kind: 'cancelled' }; + } + return { + kind: 'completed', + uploadedCount: result.uploadedCount, + piggybackedOpsCount: result.piggybackedOps.length, + localWinOpsCreated: 0, + permanentRejectionCount: 0, + hasMorePiggyback: false, + rejectedOps: [], + }; + } + if (processResult.blockedByIncompatibleOp) { return { kind: 'blocked_incompatible' }; } @@ -966,8 +1003,27 @@ export class OperationLogSyncService { result.newOps, startupCleanupFullStateOpId, startupOpIdsToDiscard, + { isNeverSynced: options?.isNeverSynced }, ); + if (processResult.preApplyFullStateConflict?.dialogData) { + const { fullStateOp, pendingOps, dialogData } = + processResult.preApplyFullStateConflict; + OpLog.warn( + `OperationLogSyncService: ${fullStateOp?.opType ?? 'Full-state op'} gained ` + + `${pendingOps.length} pending local op(s) before download apply. Showing conflict dialog.`, + ); + const conflictResult = await this._handleSyncImportConflict( + syncProvider, + dialogData, + 'OperationLogSyncService (incoming full-state pre-apply recheck)', + ); + if (conflictResult === 'CANCEL') { + return { kind: 'cancelled' }; + } + return { kind: 'no_new_ops' }; + } + if (processResult.blockedByIncompatibleOp) { return { kind: 'blocked_incompatible' }; } @@ -1054,15 +1110,53 @@ export class OperationLogSyncService { remoteOps: Operation[], fullStateOpId: string | undefined, startupOpIds: string[], - ): Promise { + conflictRecheck?: { + isNeverSynced?: boolean; + preCapturedPendingOps?: OperationLogEntry[]; + }, + ): Promise { + const startupOpIdsToDiscard = new Set(startupOpIds); + let preApplyFullStateConflict: IncomingFullStateConflictGateResult | undefined; try { - const result = await this.remoteOpsProcessingService.processRemoteOps(remoteOps); + const result = conflictRecheck + ? await this.remoteOpsProcessingService.processRemoteOps(remoteOps, { + beforeFullStateApply: async (fullStateOps): Promise => { + const conflict = + await this.syncImportConflictGateService.checkIncomingFullStateConflict( + fullStateOps, + { + isNeverSynced: conflictRecheck.isNeverSynced, + preCapturedPendingOps: conflictRecheck.preCapturedPendingOps, + }, + ); + for (const opId of conflict.discardablePendingOpIds) { + startupOpIdsToDiscard.add(opId); + } + if (conflict.dialogData) { + preApplyFullStateConflict = conflict; + return false; + } + return true; + }, + }) + : await this.remoteOpsProcessingService.processRemoteOps(remoteOps); + if ( + result.fullStateApplyBlockedByLocalConflict && + !preApplyFullStateConflict?.dialogData + ) { + throw new Error( + 'Full-state apply was blocked without conflict data for resolution.', + ); + } await this._discardStartupOpsIfFullStateCommitted( fullStateOpId, - startupOpIds, + [...startupOpIdsToDiscard], result.committedFullStateOpIds, ); - return result; + return { + ...result, + ...(preApplyFullStateConflict ? { preApplyFullStateConflict } : {}), + }; } catch (error) { try { // The reducer/apply transaction can commit the full-state op before a @@ -1070,7 +1164,7 @@ export class OperationLogSyncService { // so obsolete startup ops cannot replay after an already-applied import. await this._discardStartupOpsIfFullStateCommitted( fullStateOpId, - startupOpIds, + [...startupOpIdsToDiscard], [], true, ); 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 5848c2c44c..ec96d9615c 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 @@ -36,6 +36,8 @@ import { OpLog } from '../../core/log'; import { SyncProviderId } from '../sync-providers/provider.const'; import { LOCAL_ONLY_SYNC_KEYS } from '../../features/config/local-only-sync-settings.util'; import { IncompleteRemoteOperationsError } from '../core/errors/sync-errors'; +import { HydrationStateService } from '../apply/hydration-state.service'; +import { LOCK_NAMES } from '../core/operation-log.const'; describe('RemoteOpsProcessingService', () => { let service: RemoteOpsProcessingService; @@ -784,6 +786,15 @@ describe('RemoteOpsProcessingService', () => { schemaVersion: 1, }; const callOrder: string[] = []; + lockServiceSpy.request.and.callFake( + async (name: string, callback: () => Promise) => { + expect(name).toBe(LOCK_NAMES.UPLOAD); + callOrder.push('upload:start'); + const value = await callback(); + callOrder.push('upload:end'); + return value; + }, + ); writeFlushServiceSpy.flushPendingWrites.and.callFake(async () => { callOrder.push('flush'); }); @@ -823,6 +834,7 @@ describe('RemoteOpsProcessingService', () => { }); expect(validationSpy).toHaveBeenCalledWith(true); expect(callOrder).toEqual([ + 'upload:start', 'flush', 'exclusive:start', 'premerge', @@ -830,9 +842,58 @@ describe('RemoteOpsProcessingService', () => { 'deferred', 'validation', 'exclusive:end', + 'upload:end', ]); }); + it('should keep the final full-state guard atomic with apply and drain buffered actions on abort', async () => { + const syncImportOp: Operation = { + id: 'sync-import-final-guard', + opType: OpType.SyncImport, + actionType: '[All] Load All Data' as ActionType, + entityType: 'ALL', + payload: {}, + clientId: 'client-1', + vectorClock: { client1: 1 }, + timestamp: Date.now(), + schemaVersion: 1, + }; + const hydrationState = TestBed.inject(HydrationStateService); + const acquireHoldSpy = spyOn( + hydrationState, + 'acquireApplyingRemoteOpsHold', + ).and.callThrough(); + let wasApplyingDuringGuard = false; + let wasApplyingDuringDrain = true; + operationLogEffectsSpy.processDeferredActions.and.callFake(async () => { + wasApplyingDuringDrain = hydrationState.isApplyingRemoteOps(); + }); + + const result = await service.processRemoteOps([syncImportOp], { + beforeFullStateApply: async () => { + wasApplyingDuringGuard = hydrationState.isApplyingRemoteOps(); + return false; + }, + }); + + expect(lockServiceSpy.request).toHaveBeenCalledWith( + LOCK_NAMES.UPLOAD, + jasmine.any(Function), + ); + expect(writeFlushServiceSpy.flushThenRunExclusive).toHaveBeenCalledTimes(1); + expect(acquireHoldSpy).toHaveBeenCalledTimes(1); + expect(wasApplyingDuringGuard).toBeTrue(); + expect(operationApplierServiceSpy.applyOperations).not.toHaveBeenCalled(); + expect( + validateStateServiceSpy.validateAndRepairCurrentState, + ).not.toHaveBeenCalled(); + expect(operationLogEffectsSpy.processDeferredActions).toHaveBeenCalledOnceWith({ + callerHoldsOperationLogLock: true, + }); + expect(wasApplyingDuringDrain).toBeFalse(); + expect(result.fullStateApplyBlockedByLocalConflict).toBeTrue(); + }); + it('should propagate an already-held operation-log lock through full-state apply and validation', async () => { const syncImportOp: Operation = { id: 'sync-import-under-lock', @@ -852,13 +913,16 @@ describe('RemoteOpsProcessingService', () => { callerHoldsOperationLogLock: true, }); - expect(applySpy).toHaveBeenCalledWith([syncImportOp], true); + expect(applySpy).toHaveBeenCalledWith([syncImportOp], true, { + skipDeferredActionDrain: true, + }); expect(validationSpy).toHaveBeenCalledWith(true); expect(operationLogEffectsSpy.processDeferredActions).toHaveBeenCalledWith({ callerHoldsOperationLogLock: true, }); expect(writeFlushServiceSpy.flushThenRunExclusive).not.toHaveBeenCalled(); expect(writeFlushServiceSpy.flushPendingWrites).not.toHaveBeenCalled(); + expect(lockServiceSpy.request).not.toHaveBeenCalled(); }); it('should log incoming full-state op shape and prior receiver state for diagnostics', async () => { 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 02b52a3c1f..7055590949 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.ts @@ -37,6 +37,7 @@ import { applyLocalOnlySyncSettingsToAppData, LocalOnlySyncSettings, } from '../../features/config/local-only-sync-settings.util'; +import { HydrationStateService } from '../apply/hydration-state.service'; /** * Handles the core pipeline for processing remote operations. @@ -68,6 +69,7 @@ export class RemoteOpsProcessingService { private compactionService = inject(OperationLogCompactionService); private syncImportFilterService = inject(SyncImportFilterService); private writeFlushService = inject(OperationWriteFlushService); + private hydrationStateService = inject(HydrationStateService); private injector = inject(Injector); /** Flag to show version-incompatibility warnings only once per session */ @@ -97,6 +99,12 @@ export class RemoteOpsProcessingService { options?: { skipConflictDetection?: boolean; callerHoldsOperationLogLock?: boolean; + /** + * Final full-state conflict check. Runs with the operation-log lock held, + * immediately before the destructive reducer application. Returning false + * aborts the apply so the caller can show UI after the lock is released. + */ + beforeFullStateApply?: (fullStateOps: Operation[]) => Promise; }, ): Promise<{ localWinOpsCreated: number; @@ -111,6 +119,8 @@ export class RemoteOpsProcessingService { * when a later incompatible op blocks the remaining batch suffix. */ committedFullStateOpIds?: string[]; + /** True when beforeFullStateApply vetoed the destructive state replacement. */ + fullStateApplyBlockedByLocalConflict?: boolean; }> { // Validation failure surfaces via the SyncSessionValidationService latch // (#7330). `validateAfterSync` and the conflict-resolution validation path @@ -264,24 +274,75 @@ export class RemoteOpsProcessingService { 'RemoteOpsProcessingService: Full-state operation detected, skipping conflict detection.', ); const callerHoldsOperationLogLock = options?.callerHoldsOperationLogLock ?? false; - const applyAndValidateWithOperationLogLockHeld = async (): Promise => { - const committedFullStateOpIds = await this.applyNonConflictingOps(validOps, true); + const applyAndValidateWithOperationLogLockHeld = async (): Promise<{ + committedFullStateOpIds: string[]; + blockedByLocalConflict: boolean; + }> => { + // The operation-log lock blocks cross-tab operation writes, while this + // hold makes same-tab reducer actions enter the deferred queue. Together + // they keep the final pending-work read and destructive apply stable. + const releaseApplyingRemoteOpsHold = + this.hydrationStateService.acquireApplyingRemoteOpsHold(); + const fullStateApplyResult: { + committedFullStateOpIds: string[]; + blockedByLocalConflict: boolean; + } = { + committedFullStateOpIds: [], + blockedByLocalConflict: false, + }; + let hasPrimaryError = false; + try { + if ( + options?.beforeFullStateApply && + !(await options.beforeFullStateApply(validOps)) + ) { + fullStateApplyResult.blockedByLocalConflict = true; + } else { + fullStateApplyResult.committedFullStateOpIds = + await this.applyNonConflictingOps(validOps, true, { + skipDeferredActionDrain: true, + }); + } + } catch (error) { + hasPrimaryError = true; + throw error; + } finally { + releaseApplyingRemoteOpsHold(); + try { + await processDeferredActionsAfterRemoteApply(this.injector, true); + } catch (deferredError) { + if (!hasPrimaryError) { + throw deferredError; + } + OpLog.err( + 'RemoteOpsProcessingService: Deferred-action drain also failed after full-state processing error', + { name: (deferredError as Error | undefined)?.name }, + ); + } + } - // Clean Slate Semantics: SYNC_IMPORT/BACKUP_IMPORT replaces entire state. - // Local synced ops are NOT replayed - the import is an explicit user action - // to restore all clients to a specific point in time. - await this.validateAfterSync(true); - return committedFullStateOpIds; + if (!fullStateApplyResult.blockedByLocalConflict) { + // Clean Slate Semantics: SYNC_IMPORT/BACKUP_IMPORT replaces entire state. + // Local synced ops are NOT replayed - the import is an explicit user action + // to restore all clients to a specific point in time. Validate after the + // deferred local actions, preserving the established apply pipeline order. + await this.validateAfterSync(true); + } + return fullStateApplyResult; }; - // Keep the pending-write cutoff, remote-clock premerge, full-state reducer, - // deferred capture drain, and validation in one operation-log critical - // section. Callers already inside that section must not re-enter its - // non-reentrant lock or flush while holding it. - const committedFullStateOpIds = callerHoldsOperationLogLock + // Take UPLOAD before OPERATION_LOG, matching the established server-migration + // lock order. The outer lock prevents another tab from starting an upload + // selection/acknowledgement round during the cutoff; the inner lock keeps the + // final pending-work read, full-state reducer, deferred capture drain, and + // validation atomic with operation capture. Callers already inside OPERATION_LOG + // must not re-enter either path or flush while holding its non-reentrant lock. + const fullStateApplyResult = callerHoldsOperationLogLock ? await applyAndValidateWithOperationLogLockHeld() - : await this.writeFlushService.flushThenRunExclusive( - applyAndValidateWithOperationLogLockHeld, + : await this.lockService.request(LOCK_NAMES.UPLOAD, () => + this.writeFlushService.flushThenRunExclusive( + applyAndValidateWithOperationLogLockHeld, + ), ); return { @@ -290,7 +351,8 @@ export class RemoteOpsProcessingService { filteredOpCount: 0, isLocalUnsyncedImport: false, blockedByIncompatibleOp, - committedFullStateOpIds, + committedFullStateOpIds: fullStateApplyResult.committedFullStateOpIds, + fullStateApplyBlockedByLocalConflict: fullStateApplyResult.blockedByLocalConflict, }; } @@ -461,11 +523,14 @@ export class RemoteOpsProcessingService { * @param ops - Non-conflicting operations to apply * @param callerHoldsLock - If true, deferred local actions reuse the caller's * sp_op_log lock after remote clocks are merged. + * @param options.skipDeferredActionDrain - If true, the caller owns a wider + * apply window and drains deferred actions after validation or abort. * @throws Re-throws if application fails (ops marked as failed first) */ async applyNonConflictingOps( ops: Operation[], callerHoldsLock: boolean = false, + options: { skipDeferredActionDrain?: boolean } = {}, ): Promise { const locallyReplayableOps = await this._withLocalOnlySyncSettingsForFullStateOps(ops); @@ -550,7 +615,7 @@ export class RemoteOpsProcessingService { hasPrimaryError = true; throw error; } finally { - if (canDrainDeferredActions) { + if (canDrainDeferredActions && !options.skipDeferredActionDrain) { try { await processDeferredActionsAfterRemoteApply(this.injector, callerHoldsLock); } catch (deferredError) {