From 108b72ff1ae7ac1796afc44a605b80a08b5fd3d2 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Wed, 4 Feb 2026 19:53:44 +0100 Subject: [PATCH] fix(sync): handle MultiEntityPayload in LWW Delete vs Update conflict resolution _convertToLWWUpdatesIfNeeded() now extracts the full entity from the local DELETE payload, merges remote UPDATE changes on top, and produces a flat entity payload that lwwUpdateMetaReducer expects. Previously it only changed the actionType but kept the original MultiEntityPayload, causing the meta-reducer to bail because entityData['id'] was undefined. Also fixes _extractEntityFromDeleteOperation() to use extractActionPayload() for consistent MultiEntityPayload unwrapping, and increases E2E timing constants to ensure DELETE operations flush to IndexedDB before sync triggers. --- e2e/tests/sync/supersync-lww-conflict.spec.ts | 8 +- .../sync/conflict-resolution.service.ts | 126 ++++++++++++------ 2 files changed, 91 insertions(+), 43 deletions(-) diff --git a/e2e/tests/sync/supersync-lww-conflict.spec.ts b/e2e/tests/sync/supersync-lww-conflict.spec.ts index 818d453fdc..825718d2bc 100644 --- a/e2e/tests/sync/supersync-lww-conflict.spec.ts +++ b/e2e/tests/sync/supersync-lww-conflict.spec.ts @@ -1324,11 +1324,11 @@ test.describe('@supersync SuperSync LWW Conflict Resolution', () => { // 3. Client A deletes the task using reliable keyboard shortcut await deleteTask(clientA, taskName); - await clientA.page.waitForTimeout(300); // Flush operation to ensure DELETE is created (increased from 150ms) + await clientA.page.waitForTimeout(500); // Flush operation to ensure DELETE is created and persisted to IndexedDB console.log('[DeleteRace] Client A deleted task'); // 4. Client B updates the task (with later timestamp) - await clientB.page.waitForTimeout(1500); // Ensure UPDATE has later timestamp than DELETE (increased from 1000ms) + await clientB.page.waitForTimeout(2000); // Ensure UPDATE has later timestamp than DELETE const taskLocatorB = clientB.page .locator(`task:not(.ng-animating):has-text("${taskName}")`) @@ -1434,11 +1434,11 @@ test.describe('@supersync SuperSync LWW Conflict Resolution', () => { // 3. Client A deletes the task using reliable keyboard shortcut await deleteTask(clientA, taskName); - await clientA.page.waitForTimeout(150); // Flush operation to ensure DELETE is created and persisted + await clientA.page.waitForTimeout(500); // Flush operation to ensure DELETE is created and persisted to IndexedDB console.log('[TodayDeleteRace] Client A deleted task'); // 4. Client B updates the task (with later timestamp) - await clientB.page.waitForTimeout(1000); // Ensure later timestamp + await clientB.page.waitForTimeout(2000); // Ensure UPDATE has later timestamp than DELETE const taskLocatorB = clientB.page .locator(`task:not(.ng-animating):has-text("${taskName}")`) diff --git a/src/app/op-log/sync/conflict-resolution.service.ts b/src/app/op-log/sync/conflict-resolution.service.ts index a75df0ecb9..37f9bee2a9 100644 --- a/src/app/op-log/sync/conflict-resolution.service.ts +++ b/src/app/op-log/sync/conflict-resolution.service.ts @@ -4,6 +4,7 @@ import { ActionType, EntityConflict, EntityType, + extractActionPayload, Operation, OpType, VectorClock, @@ -18,7 +19,6 @@ import { SnackService } from '../../core/snack/snack.service'; import { T } from '../../t.const'; import { ValidateStateService } from '../validation/validate-state.service'; import { MAX_CONFLICT_RETRY_ATTEMPTS } from '../core/operation-log.const'; -import { DUPLICATE_OPERATION_ERROR_PATTERN } from '../persistence/op-log-errors.const'; import { compareVectorClocks, incrementVectorClock, @@ -817,16 +817,14 @@ export class ConflictResolutionService { return undefined; } - // Extract entity from payload based on entity type - // For TASK: payload.task - // For PROJECT: payload.project - // For TAG: payload.tag - // etc. - const payload = deleteOp.payload as Record; + // Extract entity from payload based on entity type. + // Uses extractActionPayload to handle both MultiEntityPayload format + // (where actionPayload is nested) and legacy flat payloads. + const actionPayload = extractActionPayload(deleteOp.payload); const entityKey = getPayloadKey(conflict.entityType) || conflict.entityType.toLowerCase(); - return payload[entityKey]; + return actionPayload[entityKey]; } /** @@ -851,6 +849,14 @@ export class ConflictResolutionService { return conflict.remoteOps; } + // Extract full entity from local DELETE operation's payload. + // The DELETE op carries the complete entity state at time of deletion, + // which we need as the base for merging remote UPDATE changes. + const localDeleteOp = conflict.localOps.find((op) => op.opType === OpType.Delete); + const baseEntity = localDeleteOp + ? this._extractEntityFromPayload(localDeleteOp.payload, conflict.entityType) + : undefined; + // Convert remote UPDATE operations to LWW Update format return conflict.remoteOps.map((remoteOp) => { if (remoteOp.opType === OpType.Update) { @@ -858,9 +864,30 @@ export class ConflictResolutionService { `ConflictResolutionService: Converting remote UPDATE to LWW Update for ` + `${remoteOp.entityType}:${remoteOp.entityId} (local DELETE lost)`, ); + + if (baseEntity) { + // Merge remote UPDATE changes on top of base entity from DELETE. + // This produces a flat entity payload that lwwUpdateMetaReducer can use + // to recreate the entity with all fields intact (not just the delta). + const updateChanges = this._extractUpdateChanges( + remoteOp.payload, + conflict.entityType, + ); + const mergedEntity = { ...baseEntity, ...updateChanges }; + return { + ...remoteOp, + actionType: toLwwUpdateActionType(remoteOp.entityType), + payload: mergedEntity, + }; + } + + // Fallback: can't get base entity, log warning + OpLog.warn( + `ConflictResolutionService: Cannot extract base entity from local DELETE for ` + + `${remoteOp.entityType}:${remoteOp.entityId}. LWW Update may have incomplete data.`, + ); return { ...remoteOp, - // Convert to LWW Update action type so lwwUpdateMetaReducer can recreate the entity actionType: toLwwUpdateActionType(remoteOp.entityType), }; } @@ -868,6 +895,50 @@ export class ConflictResolutionService { }); } + /** + * Extracts entity state from an operation payload. + * Handles both MultiEntityPayload format and flat payloads. + */ + private _extractEntityFromPayload( + payload: unknown, + entityType: EntityType, + ): Record | undefined { + const actionPayload = extractActionPayload(payload); + const entityKey = getPayloadKey(entityType) || entityType.toLowerCase(); + const entity = actionPayload[entityKey]; + if (entity && typeof entity === 'object') { + return entity as Record; + } + // Fallback: payload might be the entity itself (LWW Update format) + if (actionPayload && typeof actionPayload === 'object' && 'id' in actionPayload) { + return actionPayload as Record; + } + return undefined; + } + + /** + * Extracts the changed fields from an UPDATE operation payload. + * Handles NgRx entity adapter format: { task: { id, changes: {...} } } + * and flat format: { task: { id, field: value } } + */ + private _extractUpdateChanges( + payload: unknown, + entityType: EntityType, + ): Record { + const actionPayload = extractActionPayload(payload); + const entityKey = getPayloadKey(entityType) || entityType.toLowerCase(); + const entityPayload = actionPayload[entityKey] as Record | undefined; + if (!entityPayload) return {}; + // NgRx adapter format: { id, changes: {...} } + if ('changes' in entityPayload && typeof entityPayload['changes'] === 'object') { + return entityPayload['changes'] as Record; + } + // Flat format: entire object is the changes (exclude 'id') + const changes = { ...entityPayload }; + delete changes['id']; + return changes; + } + /** * Gets the current state of an entity from the NgRx store. * Uses the entity registry to look up the appropriate selector. @@ -1227,44 +1298,21 @@ export class ConflictResolutionService { } /** - * Filters out already-applied ops and appends new ones to the store, with retry on duplicate detection. - * Handles the race condition where filterNewOps uses a stale cache (issue #6213). + * Atomically filters out already-applied ops and appends new ones to the store. + * Uses appendBatchSkipDuplicates() to check and insert within a single IndexedDB + * transaction, eliminating the TOCTOU race condition (issue #6343). * * @param ops - Operations to filter and potentially append * @param source - Source of operations ('local' or 'remote') - * @param options - Options for appendBatch (e.g., pendingApply) - * @returns Object containing the filtered ops and their sequence numbers (if applicable) + * @param options - Options for appendBatchSkipDuplicates (e.g., pendingApply) + * @returns Object containing the written ops and their sequence numbers */ private async _filterAndAppendOpsWithRetry( ops: Operation[], source: 'local' | 'remote', options?: { pendingApply?: boolean }, ): Promise<{ ops: Operation[]; seqs: number[] }> { - const attemptFilterAndAppend = async (): Promise<{ - ops: Operation[]; - seqs: number[]; - }> => { - const filteredOps = await this.opLogStore.filterNewOps(ops); - if (filteredOps.length === 0) { - return { ops: [], seqs: [] }; - } - // Only pass options if defined to maintain original call signature - const seqs = options - ? await this.opLogStore.appendBatch(filteredOps, source, options) - : await this.opLogStore.appendBatch(filteredOps, source); - return { ops: filteredOps, seqs }; - }; - - try { - return await attemptFilterAndAppend(); - } catch (e) { - if (e instanceof Error && e.message.includes(DUPLICATE_OPERATION_ERROR_PATTERN)) { - OpLog.warn( - 'ConflictResolutionService: Duplicate detected, retrying with fresh filter (issue #6213 recovery)', - ); - return await attemptFilterAndAppend(); - } - throw e; - } + const result = await this.opLogStore.appendBatchSkipDuplicates(ops, source, options); + return { ops: result.writtenOps, seqs: result.seqs }; } }