From d316a684bbb6316841eaad3ba95cfa7ce99bc784 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Tue, 24 Mar 2026 14:54:27 +0100 Subject: [PATCH] fix(sync): improve move operation reliability during sync - moveItemAfterAnchor: preserve current position when anchor is concurrently deleted instead of appending to end (which corrupted ordering on remote clients). For cross-list moves, append to end as fallback to prevent data loss. - Prevent double-write of deferred actions: track buffered actions in a WeakSet and filter them in the effect so they are only written once by processDeferredActions(). - Propagate skipDequeue through handleQuotaExceeded retry path to prevent queue desync when deferred actions hit storage quota. - Add e2e tests for concurrent delete + reorder sync scenarios. --- ...upersync-concurrent-delete-reorder.spec.ts | 233 ++++++++++++++++++ .../store/work-context-meta.helper.spec.ts | 7 +- .../store/work-context-meta.helper.ts | 10 +- .../capture/operation-capture.meta-reducer.ts | 31 ++- .../op-log/capture/operation-log.effects.ts | 54 ++-- 5 files changed, 310 insertions(+), 25 deletions(-) create mode 100644 e2e/tests/sync/supersync-concurrent-delete-reorder.spec.ts diff --git a/e2e/tests/sync/supersync-concurrent-delete-reorder.spec.ts b/e2e/tests/sync/supersync-concurrent-delete-reorder.spec.ts new file mode 100644 index 0000000000..755143df10 --- /dev/null +++ b/e2e/tests/sync/supersync-concurrent-delete-reorder.spec.ts @@ -0,0 +1,233 @@ +import { test, expect } from '../../fixtures/supersync.fixture'; +import { + createTestUser, + getSuperSyncConfig, + createSimulatedClient, + closeClient, + waitForTask, + deleteTask, + getTaskTitles, + type SimulatedE2EClient, +} from '../../utils/supersync-helpers'; + +/** + * SuperSync Concurrent Delete + Reorder E2E Tests + * + * Validates that concurrent task deletion on one client and task reordering + * on another client results in consistent, correct ordering after sync. + * + * These tests cover: + * - Deferred action handling (actions buffered during sync are properly persisted) + * - Task ordering preservation when concurrent modifications overlap + * - No silent data loss (tasks not duplicated or dropped) + */ + +test.describe('@supersync Concurrent Delete and Reorder', () => { + /** + * Scenario: Client A deletes a task while Client B reorders tasks concurrently + * + * Actions: + * 1. Client A creates 4 tasks, syncs to server + * 2. Client B syncs (downloads all tasks) + * 3. Client A deletes Task2 (offline) + * 4. Client B moves Task4 up (offline, keyboard shortcut) + * 5. Client A syncs (uploads delete) + * 6. Client B syncs (downloads delete, uploads reorder) + * 7. Client A syncs (downloads reorder) + * 8. Verify: both clients have 3 tasks in consistent order, no duplicates + */ + test('Delete on one client + reorder on another converges correctly', async ({ + browser, + baseURL, + testRunId, + }) => { + const uniqueId = Date.now(); + let clientA: SimulatedE2EClient | null = null; + let clientB: SimulatedE2EClient | null = null; + + try { + const user = await createTestUser(testRunId); + const syncConfig = getSuperSyncConfig(user); + + // ============ PHASE 1: Client A Creates Tasks ============ + clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId); + await clientA.sync.setupSuperSync(syncConfig); + + const task1 = `T1-${uniqueId}`; + const task2 = `T2-${uniqueId}`; + const task3 = `T3-${uniqueId}`; + const task4 = `T4-${uniqueId}`; + + await clientA.workView.addTask(task1); + await clientA.workView.addTask(task2); + await clientA.workView.addTask(task3); + await clientA.workView.addTask(task4); + + await clientA.sync.syncAndWait(); + console.log('[ConcurrentDeleteReorder] Client A created 4 tasks and synced'); + + // ============ PHASE 2: Client B Downloads ============ + clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId); + await clientB.sync.setupSuperSync(syncConfig); + await clientB.sync.syncAndWait(); + + await waitForTask(clientB.page, task1); + await waitForTask(clientB.page, task4); + console.log('[ConcurrentDeleteReorder] Client B synced, has all 4 tasks'); + + // ============ PHASE 3: Concurrent Modifications (Offline) ============ + // Client A deletes Task2 + await deleteTask(clientA, task2); + console.log('[ConcurrentDeleteReorder] Client A deleted Task2'); + + // Client B moves Task4 up (concurrent, no sync) + const task4B = clientB.page.locator(`task:has-text("${task4}")`); + await task4B.click(); + await clientB.page.keyboard.press('Control+Shift+ArrowUp'); + await clientB.page.waitForTimeout(300); + console.log('[ConcurrentDeleteReorder] Client B moved Task4 up'); + + // ============ PHASE 4: Sync Convergence ============ + await clientA.sync.syncAndWait(); + console.log('[ConcurrentDeleteReorder] Client A synced delete'); + + await clientB.sync.syncAndWait(); + console.log( + '[ConcurrentDeleteReorder] Client B synced (download delete + upload reorder)', + ); + + await clientA.sync.syncAndWait(); + console.log('[ConcurrentDeleteReorder] Client A synced (download reorder)'); + + // Allow UI to settle + await clientA.page.waitForTimeout(500); + await clientB.page.waitForTimeout(500); + + // ============ PHASE 5: Verify Convergence ============ + const titlesA = await getTaskTitles(clientA); + const titlesB = await getTaskTitles(clientB); + + console.log('[ConcurrentDeleteReorder] Client A tasks:', titlesA); + console.log('[ConcurrentDeleteReorder] Client B tasks:', titlesB); + + // Both clients should have exactly 3 tasks (Task2 was deleted) + expect(titlesA.length).toBe(3); + expect(titlesB.length).toBe(3); + + // Both clients should have the same tasks + expect(titlesA.some((t) => t.includes('T1'))).toBe(true); + expect(titlesA.some((t) => t.includes('T3'))).toBe(true); + expect(titlesA.some((t) => t.includes('T4'))).toBe(true); + expect(titlesA.some((t) => t.includes('T2'))).toBe(false); + + // Both clients should converge to the same order + expect(titlesA).toEqual(titlesB); + + console.log('[ConcurrentDeleteReorder] Both clients converged correctly'); + } finally { + if (clientA) await closeClient(clientA); + if (clientB) await closeClient(clientB); + } + }); + + /** + * Scenario: Multiple deletes on one client, reorders on the other + * + * Stress-tests the deferred action pipeline with more operations. + * + * Actions: + * 1. Client A creates 6 tasks, syncs + * 2. Client B syncs + * 3. Client A deletes Task2 and Task4 (offline) + * 4. Client B moves Task6 to top (multiple Ctrl+Shift+Up presses) + * 5. Sync convergence + * 6. Verify: both clients have 4 tasks, consistent order + */ + test('Multiple deletes + reorder converges correctly', async ({ + browser, + baseURL, + testRunId, + }) => { + const uniqueId = Date.now(); + let clientA: SimulatedE2EClient | null = null; + let clientB: SimulatedE2EClient | null = null; + + try { + const user = await createTestUser(testRunId); + const syncConfig = getSuperSyncConfig(user); + + // ============ PHASE 1: Client A Creates Tasks ============ + clientA = await createSimulatedClient(browser, baseURL!, 'A', testRunId); + await clientA.sync.setupSuperSync(syncConfig); + + const tasks = Array.from({ length: 6 }, (_, i) => `T${i + 1}-${uniqueId}`); + for (const task of tasks) { + await clientA.workView.addTask(task); + } + + await clientA.sync.syncAndWait(); + console.log('[MultiDeleteReorder] Client A created 6 tasks and synced'); + + // ============ PHASE 2: Client B Downloads ============ + clientB = await createSimulatedClient(browser, baseURL!, 'B', testRunId); + await clientB.sync.setupSuperSync(syncConfig); + await clientB.sync.syncAndWait(); + + await waitForTask(clientB.page, tasks[0]); + await waitForTask(clientB.page, tasks[5]); + console.log('[MultiDeleteReorder] Client B synced, has all 6 tasks'); + + // ============ PHASE 3: Concurrent Modifications ============ + // Client A deletes Task2 and Task4 + await deleteTask(clientA, tasks[1]); // T2 + await deleteTask(clientA, tasks[3]); // T4 + console.log('[MultiDeleteReorder] Client A deleted T2 and T4'); + + // Client B moves Task6 towards top (3 presses) + const task6B = clientB.page.locator(`task:has-text("${tasks[5]}")`); + await task6B.click(); + for (let i = 0; i < 3; i++) { + await clientB.page.keyboard.press('Control+Shift+ArrowUp'); + await clientB.page.waitForTimeout(200); + } + console.log('[MultiDeleteReorder] Client B moved T6 up'); + + // ============ PHASE 4: Sync Convergence ============ + await clientA.sync.syncAndWait(); + await clientB.sync.syncAndWait(); + await clientA.sync.syncAndWait(); + + await clientA.page.waitForTimeout(500); + await clientB.page.waitForTimeout(500); + + // ============ PHASE 5: Verify ============ + const titlesA = await getTaskTitles(clientA); + const titlesB = await getTaskTitles(clientB); + + console.log('[MultiDeleteReorder] Client A tasks:', titlesA); + console.log('[MultiDeleteReorder] Client B tasks:', titlesB); + + // Both should have 4 tasks (T2 and T4 deleted) + expect(titlesA.length).toBe(4); + expect(titlesB.length).toBe(4); + + // Deleted tasks should be gone + expect(titlesA.some((t) => t.includes('T2'))).toBe(false); + expect(titlesA.some((t) => t.includes('T4'))).toBe(false); + + // Remaining tasks should be present + expect(titlesA.some((t) => t.includes('T1'))).toBe(true); + expect(titlesA.some((t) => t.includes('T3'))).toBe(true); + expect(titlesA.some((t) => t.includes('T5'))).toBe(true); + expect(titlesA.some((t) => t.includes('T6'))).toBe(true); + + // Both clients should converge to the same order + expect(titlesA).toEqual(titlesB); + + console.log('[MultiDeleteReorder] Both clients converged correctly'); + } finally { + if (clientA) await closeClient(clientA); + if (clientB) await closeClient(clientB); + } + }); +}); diff --git a/src/app/features/work-context/store/work-context-meta.helper.spec.ts b/src/app/features/work-context/store/work-context-meta.helper.spec.ts index 90f78c1c2f..65dc3ed51c 100644 --- a/src/app/features/work-context/store/work-context-meta.helper.spec.ts +++ b/src/app/features/work-context/store/work-context-meta.helper.spec.ts @@ -25,8 +25,13 @@ describe('moveItemAfterAnchor()', () => { expect(result).toEqual(['A', 'B', 'C', 'x']); }); - it('should handle anchor not found by appending to end', () => { + it('should preserve current position when anchor not found (no-op)', () => { const result = moveItemAfterAnchor('x', 'missing', ['A', 'x', 'B', 'C']); + expect(result).toEqual(['A', 'x', 'B', 'C']); + }); + + it('should append to end when anchor not found and item not in list (cross-list move)', () => { + const result = moveItemAfterAnchor('x', 'missing', ['A', 'B', 'C']); expect(result).toEqual(['A', 'B', 'C', 'x']); }); diff --git a/src/app/features/work-context/store/work-context-meta.helper.ts b/src/app/features/work-context/store/work-context-meta.helper.ts index d1d26863d7..0e4cb8caeb 100644 --- a/src/app/features/work-context/store/work-context-meta.helper.ts +++ b/src/app/features/work-context/store/work-context-meta.helper.ts @@ -25,8 +25,14 @@ export const moveItemAfterAnchor = ( const anchorIndex = listWithoutItem.indexOf(afterItemId); if (anchorIndex === -1) { - // Anchor not found - append to end as fallback - // This handles the case where the anchor was deleted concurrently + // Anchor not found (concurrently deleted). + if (listWithoutItem.length < currentList.length) { + // Item already in list — preserve current position (no-op). + // Returns the original array reference so NgRx selectors detect no change. + return currentList; + } + // Item is new to this list (cross-list move) — append to end as fallback + // to prevent the item from being silently dropped. return [...listWithoutItem, itemId]; } diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.ts b/src/app/op-log/capture/operation-capture.meta-reducer.ts index bb30f897e0..f2618bc4ca 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.ts @@ -79,6 +79,25 @@ let isApplyingRemoteOps = false; */ let deferredActions: PersistentAction[] = []; +/** + * Tracks actions that were deferred (buffered) during sync replay. + * The effect uses this to skip deferred actions — they will be processed + * later by processDeferredActions() with fresh vector clocks. + * + * Without this, deferred actions would be double-written: once by the effect + * (which sees them on the Actions observable) and again by processDeferredActions(). + * The effect's dequeue() would also desync the queue since nothing was enqueued. + */ +const deferredActionSet = new WeakSet(); + +/** + * Checks whether an action was deferred (buffered for post-sync processing). + * Used by the effect to skip deferred actions in the normal persistence path. + */ +export const isDeferredAction = (action: PersistentAction): boolean => { + return deferredActionSet.has(action); +}; + /** * Sets the service instance for the meta-reducer. * Must be called during app initialization before any persistent actions are dispatched. @@ -130,7 +149,11 @@ const MAX_DEFERRED_ACTIONS_HARD_LIMIT = 100; * Called by the meta-reducer when a persistent action arrives during sync. */ export const bufferDeferredAction = (action: PersistentAction): void => { - // Hard limit: drop oldest action if buffer is full (sync stuck scenario) + // Hard limit: drop oldest action if buffer is full (sync stuck scenario). + // NOTE: The shifted action remains in deferredActionSet (WeakSet has no delete-by-value). + // The effect filters it via isDeferredAction(), and getDeferredActions() won't return it, + // so it is silently lost. This is acceptable: the hard limit is itself an error condition + // (sync stuck), and dropping the oldest action is the lesser evil vs unbounded growth. if (deferredActions.length >= MAX_DEFERRED_ACTIONS_HARD_LIMIT) { devError( `[operationCaptureMetaReducer] Deferred actions buffer exceeded ${MAX_DEFERRED_ACTIONS_HARD_LIMIT} items. ` + @@ -140,6 +163,7 @@ export const bufferDeferredAction = (action: PersistentAction): void => { } deferredActions.push(action); + deferredActionSet.add(action); // Soft warning at 10 items if (deferredActions.length > MAX_DEFERRED_ACTIONS_WARNING) { @@ -162,6 +186,11 @@ export const getDeferredActions = (): PersistentAction[] => { /** * Clears the deferred actions buffer without processing. * Used for cleanup during testing or error recovery. + * + * Note: deferredActionSet (WeakSet) is not cleared here because WeakSet has + * no .clear() method. Entries are garbage-collected when action references are + * released. In practice, cleared actions are not reused by reference, so stale + * entries in the WeakSet are harmless. */ export const clearDeferredActions = (): void => { deferredActions = []; diff --git a/src/app/op-log/capture/operation-log.effects.ts b/src/app/op-log/capture/operation-log.effects.ts index 23e8ee813f..6d200ba675 100644 --- a/src/app/op-log/capture/operation-log.effects.ts +++ b/src/app/op-log/capture/operation-log.effects.ts @@ -26,7 +26,7 @@ import { import { CURRENT_SCHEMA_VERSION } from '../persistence/schema-migration.service'; import { OperationCaptureService } from './operation-capture.service'; import { ImmediateUploadService } from '../sync/immediate-upload.service'; -import { getDeferredActions } from './operation-capture.meta-reducer'; +import { getDeferredActions, isDeferredAction } from './operation-capture.meta-reducer'; import { ClientIdService } from '../../core/util/client-id.service'; import { SuperSyncStatusService } from '../sync/super-sync-status.service'; @@ -69,30 +69,34 @@ export class OperationLogEffects { * Filters out: * 1. Non-persistent actions (actions without PersistentActionMeta) * 2. Remote actions (actions replayed from sync, marked with isRemote: true) + * 3. Deferred actions (buffered during sync, processed later by processDeferredActions) * - * Note: We intentionally do NOT filter by `isApplyingRemoteOps()` here. - * The meta-reducer handles sync timing by BUFFERING actions during sync - * (not enqueueing them). If an action reaches this effect and was enqueued, - * it means the meta-reducer determined it should be processed. - * - * Previously there was a filter here that caused a race condition: - * 1. Meta-reducer enqueues action (isApplyingRemoteOps = false) - * 2. Before effect runs, sync starts (isApplyingRemoteOps = true) - * 3. Effect filters out action, but it's already in queue - * 4. flushPendingWrites() times out waiting for queue to drain + * Note: We do NOT filter by `isApplyingRemoteOps()` here because of a race + * condition: the meta-reducer may enqueue an action before sync starts, but + * the effect processes it after sync starts. Filtering by isApplyingRemoteOps + * would skip the action while its queue entry remains, causing flushPendingWrites + * to time out. Instead, we use isDeferredAction() which precisely identifies + * actions that were buffered (not enqueued) by the meta-reducer. */ persistOperation$ = createEffect( () => this.actions$.pipe( - filter((action) => isPersistentAction(action)), - filter((action) => !(action as PersistentAction).meta.isRemote), + filter( + (action): action is PersistentAction => + isPersistentAction(action) && + !action.meta.isRemote && + !isDeferredAction(action), + ), // Use concatMap for sequential processing to maintain FIFO queue order - concatMap((action) => this.writeOperation(action as PersistentAction)), + concatMap((action) => this.writeOperation(action)), ), { dispatch: false }, ); - private async writeOperation(action: PersistentAction): Promise { + private async writeOperation( + action: PersistentAction, + skipDequeue = false, + ): Promise { // Always read from ClientIdService (which has its own in-memory cache). // Do NOT cache locally — BackupService.generateNewClientId() updates the // service cache, and a local cache here would go stale after backup import. @@ -119,7 +123,9 @@ export class OperationLogEffects { ); if (!isBulkAllOperation && !hasValidEntityId && !hasValidEntityIds) { // IMPORTANT: Dequeue first to prevent queue from getting stuck - this.operationCaptureService.dequeue(); + if (!skipDequeue) { + this.operationCaptureService.dequeue(); + } devError( `[OperationLogEffects] Action ${action.type} has invalid entityId/entityIds (${action.meta.entityId}) - skipping persistence`, ); @@ -163,7 +169,9 @@ export class OperationLogEffects { // Get entity changes from the FIFO queue (for TIME_TRACKING and TASK time sync) // For most actions, this returns empty array since action payloads are sufficient. // IMPORTANT: This MUST happen inside the lock to prevent race with flushPendingWrites. - const entityChanges = this.operationCaptureService.dequeue(); + // Deferred actions skip dequeue because the meta-reducer buffers them without + // enqueueing — there's no matching queue entry to dequeue. + const entityChanges = skipDequeue ? [] : this.operationCaptureService.dequeue(); // Create multi-entity payload with action payload and computed changes const multiEntityPayload: MultiEntityPayload = { @@ -274,7 +282,7 @@ export class OperationLogEffects { ); this.notifyUserAndTriggerRollback(); } else { - await this.handleQuotaExceeded(action); + await this.handleQuotaExceeded(action, skipDequeue); } } else { this.notifyUserAndTriggerRollback(); @@ -358,7 +366,10 @@ export class OperationLogEffects { * handles quota exceeded at a time. Uses instance flag to prevent * recursive calls within the same tab. */ - private async handleQuotaExceeded(action: PersistentAction): Promise { + private async handleQuotaExceeded( + action: PersistentAction, + skipDequeue = false, + ): Promise { OpLog.err( 'OperationLogEffects: Storage quota exceeded, attempting emergency compaction', ); @@ -372,7 +383,7 @@ export class OperationLogEffects { // Set circuit breaker before retry to prevent recursive handling this.isHandlingQuotaExceeded = true; // Retry the failed operation after compaction freed space - await this.writeOperation(action); + await this.writeOperation(action, skipDequeue); this.snackService.open({ type: 'SUCCESS', msg: T.F.SYNC.S.STORAGE_RECOVERED_AFTER_COMPACTION, @@ -441,7 +452,8 @@ export class OperationLogEffects { for (let attempt = 0; attempt < MAX_RETRIES; attempt++) { try { - await this.writeOperation(action); + // skipDequeue=true: deferred actions were buffered, not enqueued + await this.writeOperation(action, true); success = true; break; } catch (e) {