mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
fix(sync): guard full-state apply against late local ops (#8976)
* fix(sync): guard full-state apply against late local ops Recheck pending work inside the upload and operation-log locks before destructive imports. Defer same-tab actions through the cutoff and keep cursors and acknowledgements unchanged when dialog resolution is required. Closes #8310 * test(sync): cover late multi-tab full-state race Use the real Web Lock boundary and shared IndexedDB to prove a sibling-tab operation blocks destructive full-state apply.
This commit is contained in:
Johannes Millan
GitHub
parent
5624f6891d
commit
6ac06df6c2
7 changed files with 707 additions and 52 deletions
|
|
@ -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<void> => {
|
||||
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<void> => {
|
||||
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<void>((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<void> => {
|
||||
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<void> => {
|
||||
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<number> =>
|
||||
page.evaluate(async () => {
|
||||
const db = await new Promise<IDBDatabase>((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<string | null> =>
|
||||
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);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<typeof setTimeout> | 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);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<boolean>;
|
||||
}
|
||||
| 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');
|
||||
|
|
|
|||
|
|
@ -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<RemoteOpsProcessingService['processRemoteOps']>
|
||||
>;
|
||||
|
||||
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<RemoteOpsProcessingResult> {
|
||||
conflictRecheck?: {
|
||||
isNeverSynced?: boolean;
|
||||
preCapturedPendingOps?: OperationLogEntry[];
|
||||
},
|
||||
): Promise<GuardedRemoteOpsProcessingResult> {
|
||||
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<boolean> => {
|
||||
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,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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 <T>(name: string, callback: () => Promise<T>) => {
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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<boolean>;
|
||||
},
|
||||
): 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<string[]> => {
|
||||
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<string[]> {
|
||||
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) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue