diff --git a/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts b/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts index ab0aca9955..cc098284d0 100644 --- a/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts +++ b/src/app/op-log/sync/conflict-journal-hook.integration.spec.ts @@ -56,6 +56,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { interface Ctx { localPendingOpsByEntity: Map; appliedFrontierByEntity: Map; + retainedOpsByEntity: Map; snapshotVectorClock: VectorClock | undefined; snapshotEntityKeys: Set | undefined; hasNoSnapshotClock: boolean; @@ -64,6 +65,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { const ctx = (over: Partial = {}): Ctx => ({ localPendingOpsByEntity: new Map(), appliedFrontierByEntity: new Map(), + retainedOpsByEntity: new Map(), snapshotVectorClock: undefined, snapshotEntityKeys: undefined, hasNoSnapshotClock: true, @@ -87,9 +89,13 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => { 'markRejected', 'markFailed', 'getUnsyncedByEntity', + 'getOpById', 'mergeRemoteOpClocks', 'markReducersCommittedAndMergeClocks', ]); + // Row lookup can't resolve in this mocked store → the synced-op rejection + // guard fails open (keeps the pending-path rejection behavior). + mockOpLogStore.getOpById.and.resolveTo(undefined); mockOpLogStore.mergeRemoteOpClocks.and.resolveTo(undefined); mockOpLogStore.markReducersCommittedAndMergeClocks.and.resolveTo(undefined); mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake(async (batches) => ({ diff --git a/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts b/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts index 3fbcf3223f..0d58c5c032 100644 --- a/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.disjoint-merge.spec.ts @@ -100,9 +100,11 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { 'markRejected', 'markFailed', 'getUnsyncedByEntity', + 'getOpById', 'mergeRemoteOpClocks', 'markReducersCommittedAndMergeClocks', ]); + mockOpLogStore.getOpById.and.resolveTo(undefined); mockOpLogStore.mergeRemoteOpClocks.and.resolveTo(undefined); mockOpLogStore.markReducersCommittedAndMergeClocks.and.resolveTo(undefined); mockOpLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake(async (batches) => ({ @@ -1364,9 +1366,11 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { 'markRejected', 'markFailed', 'getUnsyncedByEntity', + 'getOpById', 'mergeRemoteOpClocks', 'markReducersCommittedAndMergeClocks', ]); + opLogStore.getOpById.and.resolveTo(undefined); opLogStore.mergeRemoteOpClocks.and.resolveTo(undefined); opLogStore.markReducersCommittedAndMergeClocks.and.resolveTo(undefined); opLogStore.appendMixedSourceBatchSkipDuplicates.and.callFake(async (batches) => ({ diff --git a/src/app/op-log/sync/conflict-resolution.service.spec.ts b/src/app/op-log/sync/conflict-resolution.service.spec.ts index 23f055c9d5..3f1c5be9b0 100644 --- a/src/app/op-log/sync/conflict-resolution.service.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.service.spec.ts @@ -6306,18 +6306,21 @@ describe('ConflictResolutionService', () => { const buildCtx = (overrides: { localPendingOpsByEntity?: Map; appliedFrontierByEntity?: Map; + retainedOpsByEntity?: Map; snapshotVectorClock?: VectorClock; snapshotEntityKeys?: Set; hasNoSnapshotClock?: boolean; }): { localPendingOpsByEntity: Map; appliedFrontierByEntity: Map; + retainedOpsByEntity: Map; snapshotVectorClock: VectorClock | undefined; snapshotEntityKeys: Set; hasNoSnapshotClock: boolean; } => ({ localPendingOpsByEntity: overrides.localPendingOpsByEntity ?? new Map(), appliedFrontierByEntity: overrides.appliedFrontierByEntity ?? new Map(), + retainedOpsByEntity: overrides.retainedOpsByEntity ?? new Map(), snapshotVectorClock: overrides.snapshotVectorClock, snapshotEntityKeys: overrides.snapshotEntityKeys ?? new Set(), hasNoSnapshotClock: overrides.hasNoSnapshotClock ?? true, @@ -6637,6 +6640,380 @@ describe('ConflictResolutionService', () => { }); }); + describe('checkOpForConflicts — no-pending CONCURRENT crossing (#9073)', () => { + const KEY = 'TASK:task-1'; + + const buildCtx = (overrides: { + retainedOpsByEntity?: Map; + appliedFrontierByEntity?: Map; + }): { + localPendingOpsByEntity: Map; + appliedFrontierByEntity: Map; + retainedOpsByEntity: Map; + snapshotVectorClock: VectorClock | undefined; + snapshotEntityKeys: Set; + hasNoSnapshotClock: boolean; + } => ({ + localPendingOpsByEntity: new Map(), + appliedFrontierByEntity: overrides.appliedFrontierByEntity ?? new Map(), + retainedOpsByEntity: overrides.retainedOpsByEntity ?? new Map(), + snapshotVectorClock: undefined, + snapshotEntityKeys: new Set(), + hasNoSnapshotClock: true, + }); + + const updateOp = (over: { + id: string; + clientId: string; + vectorClock: VectorClock; + timestamp: number; + changes: Record; + actionType?: ActionType; + opType?: OpType; + entityIds?: string[]; + }): Operation => ({ + id: over.id, + clientId: over.clientId, + actionType: over.actionType ?? ('[Task] Update' as ActionType), + opType: over.opType ?? OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + ...(over.entityIds ? { entityIds: over.entityIds } : {}), + payload: { task: { id: 'task-1', changes: over.changes } }, + vectorClock: over.vectorClock, + timestamp: over.timestamp, + schemaVersion: 1, + }); + + /** Runs detection from one client's point of view (entity exists). */ + const detect = async ( + incoming: Operation, + retained: Operation[], + ): Promise<{ isSupersededOrDuplicate: boolean; conflicts: EntityConflict[] }> => { + mockStore.select.and.returnValue(of({ id: 'task-1', title: 'existing' })); + return service.checkOpForConflicts( + incoming, + buildCtx({ + retainedOpsByEntity: new Map([[KEY, retained]]), + appliedFrontierByEntity: new Map([ + [KEY, retained[retained.length - 1]?.vectorClock ?? {}], + ]), + }), + ); + }; + + // X and Y form a plain two-client crossing: both already synced, both + // sides pending-empty, same real field → NOT commutative. + const opX = updateOp({ + id: 'op-x', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: { title: 'from A' }, + }); + const opY = updateOp({ + id: 'op-y', + clientId: 'clientB', + vectorClock: { clientB: 1 }, + timestamp: 2000, + changes: { title: 'from B' }, + }); + + it('routes a reversed-delivery crossing into a conflict on BOTH sides (was: silent arrival-order apply)', async () => { + // Client A's view: applied+synced X, receives Y. + const onA = await detect(opY, [opX]); + // Client B's view: applied+synced Y, receives X. + const onB = await detect(opX, [opY]); + + expect(onA.isSupersededOrDuplicate).toBe(false); + expect(onB.isSupersededOrDuplicate).toBe(false); + expect(onA.conflicts.length).toBe(1); + expect(onB.conflicts.length).toBe(1); + expect(onA.conflicts[0].localOps).toEqual([opX]); + expect(onA.conflicts[0].remoteOps).toEqual([opY]); + expect(onB.conflicts[0].localOps).toEqual([opY]); + expect(onB.conflicts[0].remoteOps).toEqual([opX]); + }); + + it('resolves the mirrored conflicts to the SAME winner on both sides (convergence)', async () => { + const onA = await detect(opY, [opX]); + const onB = await detect(opX, [opY]); + + // A: remote (Y, ts 2000) wins → Y is applied via the pipeline, no local-win op. + mockStore.select.and.returnValue(of({ id: 'task-1', title: 'from A' })); + const resolutionA = await service.autoResolveConflictsLWW(onA.conflicts); + // B: local (Y) wins → ONE dominating LWW op carries B's state everywhere. + mockStore.select.and.returnValue(of({ id: 'task-1', title: 'from B' })); + const resolutionB = await service.autoResolveConflictsLWW(onB.conflicts); + + expect(resolutionA.localWinOpsCreated).toBe(0); + expect(resolutionB.localWinOpsCreated).toBe(1); + const healOp = getFirstMixedLocalOp(); + // The heal dominates BOTH sides of the crossing and preserves the + // winner's timestamp (no unfair advantage in later conflicts). + expect(compareVectorClocks(healOp.vectorClock, { clientA: 1, clientB: 1 })).toBe( + VectorClockComparison.GREATER_THAN, + ); + expect(healOp.timestamp).toBe(2000); + expect(extractActionPayload(healOp.payload as never)['title']).toBe('from B'); + }); + + it('does NOT reject the already-synced local side when the remote op wins', async () => { + const onA = await detect(opY, [opX]); + // op-x is a SYNCED row: rejecting it would erase it from frontier scans. + mockOpLogStore.getOpById.and.callFake(async (id: string) => + id === 'op-x' + ? ({ op: opX, seq: 1, source: 'local', syncedAt: 123 } as never) + : undefined, + ); + + await service.autoResolveConflictsLWW(onA.conflicts); + + const rejectedIds = mockOpLogStore.markRejected.calls.allArgs().flat(2) as string[]; + expect(rejectedIds).not.toContain('op-x'); + }); + + it('uses ALL retained concurrent ops — a newer disjoint op must not mask an older overlapping one', async () => { + // L1 changes g (ts 1000), L2 changes h (ts 1100); incoming R changes g + // (ts 900). Pairing R against L2 alone would look disjoint and slip + // through as a silent apply, losing g's newer value on this client only. + const l1 = updateOp({ + id: 'op-l1', + clientId: 'clientA', + vectorClock: { clientA: 2 }, + timestamp: 1000, + changes: { title: 'newer title' }, + }); + const l2 = updateOp({ + id: 'op-l2', + clientId: 'clientA', + vectorClock: { clientA: 3 }, + timestamp: 1100, + changes: { notes: 'unrelated' }, + }); + const remote = updateOp({ + id: 'op-r', + clientId: 'clientB', + vectorClock: { clientA: 1, clientB: 1 }, + timestamp: 900, + changes: { title: 'older title' }, + }); + + const result = await detect(remote, [l1, l2]); + + expect(result.conflicts.length).toBe(1); + expect(result.conflicts[0].localOps).toEqual([l1, l2]); + }); + + it('excludes retained ops the incoming clock already dominates', async () => { + const ancestor = updateOp({ + id: 'op-ancestor', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 500, + changes: { title: 'ancestor' }, + }); + const concurrent = updateOp({ + id: 'op-concurrent', + clientId: 'clientA', + vectorClock: { clientA: 2 }, + timestamp: 1000, + changes: { title: 'concurrent' }, + }); + // Incoming saw the ancestor (clientA: 1) but not the concurrent op. + const remote = updateOp({ + id: 'op-r', + clientId: 'clientB', + vectorClock: { clientA: 1, clientB: 1 }, + timestamp: 2000, + changes: { title: 'remote' }, + }); + + const result = await detect(remote, [ancestor, concurrent]); + + expect(result.conflicts.length).toBe(1); + expect(result.conflicts[0].localOps).toEqual([concurrent]); + }); + + it('applies as-is when the retained concurrent history was compacted away (status quo)', async () => { + // Frontier still says CONCURRENT (snapshot clock), but the concurrent + // local ops themselves are gone — no deterministic local side. + mockStore.select.and.returnValue(of({ id: 'task-1', title: 'existing' })); + + const result = await service.checkOpForConflicts( + opY, + buildCtx({ + retainedOpsByEntity: new Map(), + appliedFrontierByEntity: new Map([[KEY, { clientA: 5 }]]), + }), + ); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('applies disjoint real-field crossings as-is (lossless commute preserved)', async () => { + const localNotes = updateOp({ + id: 'op-notes', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: { notes: 'local notes' }, + }); + + const result = await detect(opY, [localNotes]); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('applies identical-content crossings as-is (also damps heal-op echoes)', async () => { + const sameContent = updateOp({ + id: 'op-same', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: { title: 'from B' }, + }); + const incoming = updateOp({ + id: 'op-same-remote', + clientId: 'clientB', + vectorClock: { clientB: 1 }, + timestamp: 2000, + changes: { title: 'from B' }, + }); + + const result = await detect(incoming, [sameContent]); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('applies noise-only crossings as-is instead of minting whole-entity heals', async () => { + const noiseOnly = updateOp({ + id: 'op-noise', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 3000, + changes: { modified: 3000 }, + }); + + // Even though the noise side has the LATER timestamp, the real edit must + // not be clobbered by a whole-entity LWW win of a `modified` bump. + const result = await detect(opY, [noiseOnly]); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('keeps concurrent task-time deltas non-conflicting (commute, mirror of pending-path exemption)', async () => { + const timeOp = (id: string, clientId: string, clock: VectorClock): Operation => ({ + ...updateOp({ + id, + clientId, + vectorClock: clock, + timestamp: 1000, + changes: {}, + }), + actionType: ActionType.TIME_TRACKING_SYNC_TIME_SPENT, + payload: { taskId: 'task-1', date: '2024-01-15', duration: 2000 }, + }); + + const result = await detect(timeOp('op-time-r', 'clientB', { clientB: 1 }), [ + timeOp('op-time-l', 'clientA', { clientA: 1 }), + ]); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('applies multi-entity remote ops as-is (needs the pending path compensation machinery)', async () => { + const multiRemote = updateOp({ + id: 'op-multi', + clientId: 'clientB', + vectorClock: { clientB: 1 }, + timestamp: 2000, + changes: { title: 'bulk' }, + entityIds: ['task-1', 'task-2'], + }); + + const result = await detect(multiRemote, [opX]); + + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + }); + + it('applies as-is when the local side contains a Delete, archive, or multi-entity op', async () => { + const localDelete = updateOp({ + id: 'op-del', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: {}, + opType: OpType.Delete, + }); + const localArchive = updateOp({ + id: 'op-arch', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: {}, + actionType: ActionType.TASK_SHARED_MOVE_TO_ARCHIVE, + }); + const localMulti = updateOp({ + id: 'op-lmulti', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: { title: 'bulk' }, + entityIds: ['task-1', 'task-2'], + }); + + for (const local of [localDelete, localArchive, localMulti]) { + const result = await detect(opY, [local]); + expect(result).toEqual({ isSupersededOrDuplicate: false, conflicts: [] }); + } + }); + + it('still skips the remote op when the entity is absent (archive/delete wins, unchanged)', async () => { + mockStore.select.and.returnValue(of(undefined)); + + const result = await service.checkOpForConflicts( + opY, + buildCtx({ + retainedOpsByEntity: new Map([[KEY, [opX]]]), + appliedFrontierByEntity: new Map([[KEY, opX.vectorClock]]), + }), + ); + + expect(result.isSupersededOrDuplicate).toBe(true); + expect(result.conflicts).toEqual([]); + }); + + it('ties on timestamp resolve by clientId, symmetrically', async () => { + const tieX = updateOp({ + id: 'op-tie-x', + clientId: 'clientA', + vectorClock: { clientA: 1 }, + timestamp: 1000, + changes: { title: 'from A' }, + }); + const tieY = updateOp({ + id: 'op-tie-y', + clientId: 'clientB', + vectorClock: { clientB: 1 }, + timestamp: 1000, + changes: { title: 'from B' }, + }); + + const onA = await detect(tieY, [tieX]); + const onB = await detect(tieX, [tieY]); + + // 'clientB' > 'clientA' → Y wins on BOTH sides: remote-wins on A + // (no local-win op), local-wins on B (one heal op). + mockStore.select.and.returnValue(of({ id: 'task-1', title: 'x' })); + const resolutionA = await service.autoResolveConflictsLWW(onA.conflicts); + const resolutionB = await service.autoResolveConflictsLWW(onB.conflicts); + + expect(resolutionA.localWinOpsCreated).toBe(0); + expect(resolutionB.localWinOpsCreated).toBe(1); + }); + }); + describe('_convertToLWWUpdatesIfNeeded', () => { const createOpWithTimestamp = ( id: string, diff --git a/src/app/op-log/sync/conflict-resolution.service.ts b/src/app/op-log/sync/conflict-resolution.service.ts index 6178f1d995..df17857e20 100644 --- a/src/app/op-log/sync/conflict-resolution.service.ts +++ b/src/app/op-log/sync/conflict-resolution.service.ts @@ -84,6 +84,7 @@ import { mergeChangedFields, synthesizeMergedChanges, } from './conflict-disjoint-merge.util'; +import { NOISE_FIELDS } from './conflict-journal.model'; import { RECREATE_FALLBACK } from '../core/recreate-fallback.const'; /** @@ -1711,8 +1712,8 @@ export class ConflictResolutionService { ...merged.conflict.remoteOps.map((op) => op.id), ]), ); - const remainingLocalOpsToReject = localOpsToReject.filter( - (opId) => !fallbackOriginalOpIds.has(opId), + const remainingLocalOpsToReject = await this._withoutSyncedOps( + localOpsToReject.filter((opId) => !fallbackOriginalOpIds.has(opId)), ); const remainingRemoteOpsToReject = remoteOpsToReject.filter( (opId) => !fallbackOriginalOpIds.has(opId), @@ -3679,6 +3680,7 @@ export class ConflictResolutionService { ctx: { localPendingOpsByEntity: Map; appliedFrontierByEntity: Map; + retainedOpsByEntity: Map; snapshotVectorClock: VectorClock | undefined; snapshotEntityKeys: Set | undefined; hasNoSnapshotClock: boolean; @@ -3694,6 +3696,7 @@ export class ConflictResolutionService { const result = await this._checkEntityForConflict(remoteOp, entityId, entityKey, { localOpsForEntity, appliedFrontier: ctx.appliedFrontierByEntity.get(entityKey), + retainedOpsForEntity: ctx.retainedOpsByEntity.get(entityKey) ?? [], snapshotVectorClock: ctx.snapshotVectorClock, snapshotEntityKeys: ctx.snapshotEntityKeys, hasNoSnapshotClock: ctx.hasNoSnapshotClock, @@ -3722,6 +3725,7 @@ export class ConflictResolutionService { ctx: { localOpsForEntity: Operation[]; appliedFrontier: VectorClock | undefined; + retainedOpsForEntity: Operation[]; snapshotVectorClock: VectorClock | undefined; snapshotEntityKeys: Set | undefined; hasNoSnapshotClock: boolean; @@ -3784,6 +3788,22 @@ export class ConflictResolutionService { ); return { isSupersededOrDuplicate: true, conflict: null }; } + // #9073: the entity still exists, so a blind apply would let ARRIVAL + // ORDER decide the winner — two clients that each already synced one + // side of the crossing would keep the other's value and permanently + // diverge. Reconstruct the local side from the retained (already + // applied) concurrent ops and route it through the normal LWW + // pipeline, which resolves the same unordered op pair identically on + // every client. Crossings that commute (identical, disjoint real + // fields, noise-only, task-time deltas) keep today's lossless apply. + const crossingConflict = this._buildNoPendingConcurrentConflict( + remoteOp, + entityId, + ctx.retainedOpsForEntity, + ); + if (crossingConflict) { + return { isSupersededOrDuplicate: false, conflict: crossingConflict }; + } } return { isSupersededOrDuplicate: false, conflict: null }; } @@ -3818,6 +3838,142 @@ export class ConflictResolutionService { return { isSupersededOrDuplicate: false, conflict: null }; } + /** + * #9073: builds a synthetic conflict for a CONCURRENT remote op on an entity + * that still exists and has NO pending local ops. The local side is every + * retained op still concurrent with the incoming clock — the whole crossing, + * not just the frontier op, so a newer disjoint edit cannot mask an older + * overlapping one. Both clients reconstruct the same unordered pair, so the + * LWW plan (timestamps, then clientId) picks the same winner everywhere; a + * local win emits the usual dominating LWW Update op that heals clients + * whose frontier could not see the crossing (e.g. snapshot-clock frontiers). + * + * Returns null for crossings where applying the op as-is stays correct: + * - commuting pairs (identical content, disjoint real fields, noise-only + * sides, concurrent task-time deltas) — apply-both is lossless AND + * convergent, whole-entity LWW would discard one side; + * - cases with no deterministic local side (retained ops compacted away) + * or that would need the pending path's rejection/compensation machinery + * (multi-entity ops, local Delete/archive against a live entity) — these + * keep today's arrival-order behavior as a documented residual. + */ + private _buildNoPendingConcurrentConflict( + remoteOp: Operation, + entityId: string, + retainedOpsForEntity: Operation[], + ): EntityConflict | null { + // Resolving one entity of an atomic multi-entity op would drop its + // sibling changes without the pending path's compensation machinery. + if (getOpEntityIds(remoteOp).length > 1) { + return null; + } + + const localOps = retainedOpsForEntity.filter( + (op) => + compareVectorClocks(op.vectorClock, remoteOp.vectorClock) === + VectorClockComparison.CONCURRENT, + ); + // Empty = the CONCURRENT verdict came from the snapshot clock alone (the + // concurrent local history was compacted away) — no local side to compare. + if (localOps.length === 0) { + return null; + } + + // A local Delete/archive op with the entity still in state is a + // contradictory edge (also covers the plan's delete-win/archive-win + // kinds); multi-entity local ops would mint per-entity against an atomic + // op. Both keep the status quo. + if ( + localOps.some( + (op) => + op.opType === OpType.Delete || + op.actionType === ActionType.TASK_SHARED_MOVE_TO_ARCHIVE || + getOpEntityIds(op).length > 1, + ) + ) { + return null; + } + + // Concurrent task-time batches are positive deltas and commute — LWW + // would discard one user's tracked time (mirror of the pending-path + // exemption above). + if ( + remoteOp.actionType === ActionType.TIME_TRACKING_SYNC_TIME_SPENT && + localOps.every((op) => op.actionType === ActionType.TIME_TRACKING_SYNC_TIME_SPENT) + ) { + return null; + } + + const conflict: EntityConflict = { + entityType: remoteOp.entityType, + entityId, + localOps, + remoteOps: [remoteOp], + suggestedResolution: this._suggestResolution(localOps, [remoteOp]), + }; + + // Same content on both sides: applying is equivalent either way. This also + // damps echo rounds when several holders of the winner mint + // identical-content resolution ops that later cross each other. + if (this.isIdenticalConflict(conflict)) { + return null; + } + + const payloadKey = this._resolvePayloadKey(remoteOp.entityType); + // A side that changed nothing real: apply-both already converges on every + // real field; only noise-field arrival divergence remains (status quo, + // cosmetic). Whole-entity LWW could instead clobber the real side. + if ( + this._isNoiseOnlySide(localOps, payloadKey, entityId) || + this._isNoiseOnlySide([remoteOp], payloadKey, entityId) + ) { + return null; + } + + // Disjoint real-field updates commute — apply-both is lossless and + // convergent, while a whole-entity LWW winner would discard the loser's + // fields fleet-wide. Same predicate as the SPAP-14 merge eligibility, so a + // conflict forwarded from here is by construction never merge-eligible. + if ( + isDisjointMergeEligible({ + localOps, + remoteOps: [remoteOp], + payloadKey, + entityId, + }) + ) { + return null; + } + + OpLog.normal( + `ConflictResolutionService: No-pending CONCURRENT crossing for ` + + `${remoteOp.entityType}:${entityId} (${localOps.length} retained local op(s) ` + + `vs remote op ${remoteOp.id}) — routing through LWW (#9073)`, + ); + return conflict; + } + + /** + * True when every field the side changed is a NOISE field (and the side is + * decomposable at all — opaque ops carry real, non-extractable mutations). + */ + private _isNoiseOnlySide( + ops: Operation[], + payloadKey: string, + entityId: string, + ): boolean { + if (ops.some((op) => op.opType === OpType.Delete)) { + return false; + } + if (hasOpaqueChanges(ops, payloadKey, entityId)) { + return false; + } + const changedFields = Object.keys(mergeChangedFields(ops, payloadKey, entityId)); + return ( + changedFields.length > 0 && changedFields.every((field) => NOISE_FIELDS.has(field)) + ); + } + /** * Builds the local frontier vector clock for an entity. * Merges applied frontier + pending ops clocks. @@ -3894,6 +4050,33 @@ export class ConflictResolutionService { return suggestConflictResolution(localOps, remoteOps); } + /** + * Invariant guard: NEVER reject an already-synced op. A synced row is on the + * server and in other clients' histories; rejecting it here would erase it + * from this client's frontier scans (they skip `rejectedAt`) while the rest + * of the fleet keeps it — corrupting later conflict detection. Pending-path + * conflicts only ever queue unsynced ids (they come from getUnsyncedByEntity + * by construction), so this filter is a no-op for them; the synthetic + * no-pending crossings (#9073) queue their already-synced localOps, which + * are dropped here — the winning LWW op supersedes them by clock domination + * instead. Fails open when a row cannot be loaded. + */ + private async _withoutSyncedOps(opIds: string[]): Promise { + const result: string[] = []; + for (const opId of opIds) { + const entry = await this.opLogStore.getOpById(opId); + if (entry?.syncedAt !== undefined) { + OpLog.verbose( + `ConflictResolutionService: Not rejecting already-synced op ${opId} ` + + `(superseded by resolution op's clock instead)`, + ); + continue; + } + result.push(opId); + } + return result; + } + /** * Atomically filters out already-applied ops and appends new ones to the store. * Uses appendBatchSkipDuplicates() to check and insert within a single IndexedDB 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 3f4814c4e2..c7ce90068b 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 @@ -127,10 +127,18 @@ describe('RemoteOpsProcessingService', () => { opLogStoreSpy.clearFullStateOpsExcept.and.resolveTo(0); vectorClockServiceSpy = jasmine.createSpyObj('VectorClockService', [ 'getEntityFrontier', + 'getEntityFrontierWithOps', 'getSnapshotVectorClock', 'getSnapshotEntityKeys', 'getCurrentVectorClock', ]); + // Delegate to the getEntityFrontier spy so the many per-test frontier + // stubs keep driving the flow; retained ops default to empty (no + // no-pending crossing detection in these tests). + vectorClockServiceSpy.getEntityFrontierWithOps.and.callFake(async () => ({ + frontier: await vectorClockServiceSpy.getEntityFrontier(), + retainedOpsByEntity: new Map(), + })); operationApplierServiceSpy = jasmine.createSpyObj('OperationApplierService', [ 'applyOperations', ]); @@ -790,6 +798,7 @@ describe('RemoteOpsProcessingService', () => { expect(service.detectConflicts).toHaveBeenCalledWith( [migratedOp], jasmine.any(Map), + jasmine.any(Map), ); }); @@ -1550,7 +1559,7 @@ describe('RemoteOpsProcessingService', () => { }), ]; - const result = await service.detectConflicts(remoteTaskOps, new Map()); + const result = await service.detectConflicts(remoteTaskOps, new Map(), new Map()); // TASK op should be non-conflicting (not a conflict!) expect(result.nonConflicting.length).toBe(1); @@ -1587,7 +1596,7 @@ describe('RemoteOpsProcessingService', () => { }), ]; - const result = await service.detectConflicts(remoteOps, new Map()); + const result = await service.detectConflicts(remoteOps, new Map(), new Map()); // Should be detected as conflict (concurrent modifications to same entity) expect(result.conflicts.length).toBe(1); @@ -1622,7 +1631,7 @@ describe('RemoteOpsProcessingService', () => { conflicts, }); - const result = await service.detectConflicts([remoteOp], new Map()); + const result = await service.detectConflicts([remoteOp], new Map(), new Map()); expect(result.conflicts.map((conflict) => conflict.entityId)).toEqual([ 'task-1', @@ -1652,7 +1661,7 @@ describe('RemoteOpsProcessingService', () => { }), ]; - const result = await service.detectConflicts(remoteOps, new Map()); + const result = await service.detectConflicts(remoteOps, new Map(), new Map()); // Should be skipped (superseded) expect(result.nonConflicting.length).toBe(0); 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 51c3e137a5..3af2d9e19c 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.ts @@ -408,10 +408,12 @@ export class RemoteOpsProcessingService { // detect conflicts, AND apply resolutions. let localWinOpsCreated = 0; await this.lockService.request(LOCK_NAMES.OPERATION_LOG, async () => { - const appliedFrontierByEntity = await this.vectorClockService.getEntityFrontier(); + const { frontier: appliedFrontierByEntity, retainedOpsByEntity } = + await this.vectorClockService.getEntityFrontierWithOps(); const conflictResult = await this.detectConflicts( validOps, appliedFrontierByEntity, + retainedOpsByEntity, ); const { nonConflicting, conflicts } = conflictResult; @@ -760,11 +762,15 @@ export class RemoteOpsProcessingService { * * @param remoteOps - Remote operations to check for conflicts * @param appliedFrontierByEntity - Per-entity vector clocks of applied ops + * @param retainedOpsByEntity - Per-entity retained (non-rejected) ops from the + * same scan as the frontier; reconstructs the local side of no-pending + * CONCURRENT crossings (#9073) * @returns Object with `nonConflicting` ops to apply and `conflicts` to resolve */ async detectConflicts( remoteOps: Operation[], appliedFrontierByEntity: Map, + retainedOpsByEntity: Map, ): Promise { const localPendingOpsByEntity = await this.opLogStore.getUnsyncedByEntity(); const conflicts: EntityConflict[] = []; @@ -788,6 +794,7 @@ export class RemoteOpsProcessingService { const result = await this.conflictResolutionService.checkOpForConflicts(remoteOp, { localPendingOpsByEntity, appliedFrontierByEntity, + retainedOpsByEntity, snapshotVectorClock, snapshotEntityKeys, hasNoSnapshotClock, diff --git a/src/app/op-log/sync/vector-clock.service.spec.ts b/src/app/op-log/sync/vector-clock.service.spec.ts index 970d331677..d23b1c4022 100644 --- a/src/app/op-log/sync/vector-clock.service.spec.ts +++ b/src/app/op-log/sync/vector-clock.service.spec.ts @@ -532,4 +532,77 @@ describe('VectorClockService', () => { expect(frontier.size).toBe(0); }); }); + + describe('getEntityFrontierWithOps', () => { + it('should return a frontier identical to getEntityFrontier plus every retained op per entity', async () => { + mockStoreService.loadStateCache.and.returnValue(Promise.resolve(null)); + + const op1 = createMockOperation('op1', { clientA: 1 }, 'TASK', 'task-1'); + const op2 = createMockOperation('op2', { clientA: 2 }, 'TASK', 'task-1'); + const op3 = createMockOperation('op3', { clientB: 1 }, 'TASK', 'task-2'); + const ops: OperationLogEntry[] = [ + createMockEntry(1, op1), + createMockEntry(2, op2), + createMockEntry(3, op3), + ]; + mockStoreService.getOpsAfterSeq.and.returnValue(Promise.resolve(ops)); + + const { frontier, retainedOpsByEntity } = await service.getEntityFrontierWithOps(); + + expect(frontier).toEqual(await service.getEntityFrontier()); + expect(frontier.get('TASK:task-1')).toEqual({ clientA: 2 }); + expect(retainedOpsByEntity.get('TASK:task-1')).toEqual([op1, op2]); + expect(retainedOpsByEntity.get('TASK:task-2')).toEqual([op3]); + }); + + it('should skip rejected ops in BOTH maps (filter-identical to the frontier)', async () => { + mockStoreService.loadStateCache.and.returnValue(Promise.resolve(null)); + + const keptOp = createMockOperation('op1', { clientA: 1 }, 'TASK', 'task-1'); + const rejectedOp = createMockOperation('op2', { clientA: 2 }, 'TASK', 'task-1'); + const ops: OperationLogEntry[] = [ + createMockEntry(1, keptOp), + { ...createMockEntry(2, rejectedOp), rejectedAt: Date.now() }, + ]; + mockStoreService.getOpsAfterSeq.and.returnValue(Promise.resolve(ops)); + + const { frontier, retainedOpsByEntity } = await service.getEntityFrontierWithOps(); + + expect(frontier.get('TASK:task-1')).toEqual({ clientA: 1 }); + expect(retainedOpsByEntity.get('TASK:task-1')).toEqual([keptOp]); + }); + + it('should record a multi-entity op under every entity it touches', async () => { + mockStoreService.loadStateCache.and.returnValue(Promise.resolve(null)); + + const bulkOp = createMockOperation('op1', { clientA: 1 }, 'TASK', 'task-1', [ + 'task-1', + 'task-2', + ]); + mockStoreService.getOpsAfterSeq.and.returnValue( + Promise.resolve([createMockEntry(1, bulkOp)]), + ); + + const { retainedOpsByEntity } = await service.getEntityFrontierWithOps(); + + expect(retainedOpsByEntity.get('TASK:task-1')).toEqual([bulkOp]); + expect(retainedOpsByEntity.get('TASK:task-2')).toEqual([bulkOp]); + }); + + it('should scan from the snapshot lastAppliedOpSeq', async () => { + mockStoreService.loadStateCache.and.returnValue( + Promise.resolve({ + vectorClock: { clientA: 10 }, + lastAppliedOpSeq: 50, + state: {}, + compactedAt: Date.now(), + }), + ); + mockStoreService.getOpsAfterSeq.and.returnValue(Promise.resolve([])); + + await service.getEntityFrontierWithOps(); + + expect(mockStoreService.getOpsAfterSeq).toHaveBeenCalledWith(50); + }); + }); }); diff --git a/src/app/op-log/sync/vector-clock.service.ts b/src/app/op-log/sync/vector-clock.service.ts index dfaac57e19..ed375b3ce7 100644 --- a/src/app/op-log/sync/vector-clock.service.ts +++ b/src/app/op-log/sync/vector-clock.service.ts @@ -1,6 +1,6 @@ import { inject, Injectable } from '@angular/core'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; -import { VectorClock, EntityType } from '../core/operation.types'; +import { VectorClock, EntityType, Operation } from '../core/operation.types'; import { mergeVectorClocks } from '../../core/util/vector-clock'; import { toEntityKey } from '../util/entity-key.util'; import { getOpEntityIds } from '../util/get-op-entity-ids.util'; @@ -195,4 +195,44 @@ export class VectorClockService { return map; } + + /** + * Gets the entity frontier together with every retained (non-rejected) op + * per entity, from ONE scan of the post-snapshot log. + * + * The retained ops let conflict detection reconstruct the LOCAL side of a + * conflict for entities with no pending ops (#9073): an incoming remote op + * is concurrent with the entity's whole retained history, not just the + * frontier op, so detection must not let a newer disjoint op mask an older + * overlapping one. Both maps MUST stay filter-identical (skip `rejectedAt` + * rows only) or the frontier's CONCURRENT verdict and the reconstructed + * conflict side would disagree. + */ + async getEntityFrontierWithOps(): Promise<{ + frontier: Map; + retainedOpsByEntity: Map; + }> { + const frontier = new Map(); + const retainedOpsByEntity = new Map(); + + const snapshot = await this.opLogStore.loadStateCache(); + const ops = await this.opLogStore.getOpsAfterSeq(snapshot?.lastAppliedOpSeq || 0); + + for (const entry of ops) { + if (entry.rejectedAt) continue; + + for (const id of getOpEntityIds(entry.op)) { + const key = toEntityKey(entry.op.entityType, id); + frontier.set(key, entry.op.vectorClock); + const retained = retainedOpsByEntity.get(key); + if (retained) { + retained.push(entry.op); + } else { + retainedOpsByEntity.set(key, [entry.op]); + } + } + } + + return { frontier, retainedOpsByEntity }; + } } diff --git a/src/app/op-log/testing/integration/no-pending-crossing-convergence.integration.spec.ts b/src/app/op-log/testing/integration/no-pending-crossing-convergence.integration.spec.ts new file mode 100644 index 0000000000..171a7a031b --- /dev/null +++ b/src/app/op-log/testing/integration/no-pending-crossing-convergence.integration.spec.ts @@ -0,0 +1,340 @@ +import { TestBed } from '@angular/core/testing'; +import { Action, ActionReducer, createSelector, Store } from '@ngrx/store'; +import { of } from 'rxjs'; +import { ConflictResolutionService } from '../../sync/conflict-resolution.service'; +import { VectorClockService } from '../../sync/vector-clock.service'; +import { OperationLogStoreService } from '../../persistence/operation-log-store.service'; +import { OperationApplierService } from '../../apply/operation-applier.service'; +import { OperationCaptureService } from '../../capture/operation-capture.service'; +import { OperationLogEffects } from '../../capture/operation-log.effects'; +import { ValidateStateService } from '../../validation/validate-state.service'; +import { SnackService } from '../../../core/snack/snack.service'; +import { CLIENT_ID_PROVIDER } from '../../util/client-id.provider'; +import { buildEntityRegistry, ENTITY_REGISTRY } from '../../core/entity-registry'; +import { PersistentAction } from '../../core/persistent-action.interface'; +import { + EntityConflict, + extractActionPayload, + isLwwUpdatePayload, + Operation, +} from '../../core/operation.types'; +import { convertOpToAction } from '../../apply/operation-converter.util'; +import { + taskReducer, + TASK_FEATURE_NAME, +} from '../../../features/tasks/store/task.reducer'; +import { TaskSharedActions } from '../../../root-store/meta/task-shared.actions'; +import { Task } from '../../../features/tasks/task.model'; +import { RootState } from '../../../root-store/root-state'; +import { createStateWithExistingTasks } from '../../../root-store/meta/task-shared-meta-reducers/test-utils'; +import { + createCombinedTaskSharedMetaReducer, + updateTaskEntity, +} from '../../../root-store/meta/task-shared-meta-reducers/test-helpers'; +import { lwwUpdateMetaReducer } from '../../../root-store/meta/task-shared-meta-reducers/lww-update.meta-reducer'; +import { + compareVectorClocks, + VectorClockComparison, +} from '../../../core/util/vector-clock'; +import { resetTestUuidCounter, TestClient } from './helpers/test-client.helper'; + +/** + * #9073 — no-pending CONCURRENT crossing convergence, through the production + * wiring: real IndexedDB op-log rows, the real `getEntityFrontierWithOps` scan, + * real conflict detection, and `autoResolveConflictsLWW` with the SAME freeze + * flags the production caller passes (remote-ops-processing STEP 5). + * + * Scenario: clients A and B concurrently edit the same field of one task; both + * ops are already synced on their author before either sees the other's + * (reachable via composed third-device crossings and upload-guard escape + * hatches). Pre-#9073 both sides silently applied by arrival order and diverged + * permanently. + */ +describe('no-pending CONCURRENT crossing convergence integration (#9073)', () => { + const TASK_ID = 'task-x'; + const CLIENT_A = 'crossing-client-a'; + const CLIENT_B = 'crossing-client-b'; + const FREEZE_FLAGS = { + disableDisjointMerge: true, + disableConflictJournal: true, + } as const; + + let opLogStore: OperationLogStoreService; + let vectorClock: VectorClockService; + let resolver: ConflictResolutionService; + let capture: OperationCaptureService; + let initialState: RootState; + let localState: RootState; + let reducer: ActionReducer; + let currentClientId: string; + + const captureOperation = ( + action: PersistentAction, + client: TestClient, + timestamp: number, + ): Operation => { + const { type, meta, ...actionPayload } = action; + const entityIds = meta.entityIds ?? (meta.entityId ? [meta.entityId] : undefined); + const entityId = meta.entityId ?? entityIds?.[0]; + if (!entityId) { + throw new Error('Persistent test action has no entity id'); + } + return { + ...client.createOperation({ + actionType: type, + opType: meta.opType, + entityType: meta.entityType, + entityId, + entityIds, + payload: { + actionPayload, + entityChanges: capture.extractEntityChanges(action), + }, + }), + timestamp, + }; + }; + + const titleUpdateOp = ( + client: TestClient, + title: string, + timestamp: number, + ): Operation => + captureOperation( + TaskSharedActions.updateTask({ + task: { id: TASK_ID, changes: { title } }, + }) as PersistentAction, + client, + timestamp, + ); + + const getTitle = (): string => + (localState[TASK_FEATURE_NAME].entities[TASK_ID] as Task).title; + + /** Production-shaped detection ctx (mirrors detectConflicts' internals). */ + const detect = async ( + incoming: Operation, + ): Promise<{ isSupersededOrDuplicate: boolean; conflicts: EntityConflict[] }> => { + const { frontier, retainedOpsByEntity } = + await vectorClock.getEntityFrontierWithOps(); + return resolver.checkOpForConflicts(incoming, { + localPendingOpsByEntity: await opLogStore.getUnsyncedByEntity(), + appliedFrontierByEntity: frontier, + retainedOpsByEntity, + snapshotVectorClock: await vectorClock.getSnapshotVectorClock(), + snapshotEntityKeys: await vectorClock.getSnapshotEntityKeys(), + hasNoSnapshotClock: true, + }); + }; + + /** Installs one client's world: the given op applied to state + synced row. */ + const installSyncedLocalOp = async (op: Operation, title: string): Promise => { + await opLogStore._clearAllDataForTesting(); + localState = updateTaskEntity(initialState, TASK_ID, { title }); + const seq = await opLogStore.append(op, 'local'); + await opLogStore.markSynced([seq]); + }; + + beforeEach(async () => { + resetTestUuidCounter(); + + initialState = createStateWithExistingTasks([TASK_ID]); + initialState = updateTaskEntity(initialState, TASK_ID, { title: 'original' }); + + const rootReducer: ActionReducer = ( + state = initialState, + action, + ) => ({ + ...state, + [TASK_FEATURE_NAME]: taskReducer(state[TASK_FEATURE_NAME], action), + }); + reducer = createCombinedTaskSharedMetaReducer( + lwwUpdateMetaReducer(rootReducer), + ) as ActionReducer; + localState = initialState; + currentClientId = CLIENT_A; + + const storeSpy = jasmine.createSpyObj('Store', ['select']); + storeSpy.select.and.callFake((selector: unknown, props?: unknown) => { + if (typeof selector !== 'function') { + return of(undefined) as ReturnType; + } + const selected = ( + selector as (state: RootState, selectorProps?: unknown) => unknown + )(localState, props); + return of(selected) as ReturnType; + }); + + const applierSpy = jasmine.createSpyObj( + 'OperationApplierService', + ['applyOperations'], + ); + applierSpy.applyOperations.and.callFake(async (ops, options) => { + for (const op of ops) { + localState = reducer(localState, convertOpToAction(op)); + } + await options?.onReducersCommitted?.(ops); + return { appliedOps: ops }; + }); + + const validateSpy = jasmine.createSpyObj( + 'ValidateStateService', + ['validateAndRepairCurrentState'], + ); + validateSpy.validateAndRepairCurrentState.and.resolveTo(true); + + const effectsSpy = jasmine.createSpyObj('OperationLogEffects', [ + 'processDeferredActions', + ]); + effectsSpy.processDeferredActions.and.resolveTo(); + + const entityRegistry = buildEntityRegistry(); + const taskConfig = entityRegistry.TASK; + if (!taskConfig) { + throw new Error('TASK entity config is required for this integration test.'); + } + taskConfig.selectById = createSelector( + (state: RootState) => state[TASK_FEATURE_NAME], + (state, props: { id: string }) => state.entities[props.id] as Task, + ) as unknown as NonNullable; + + TestBed.configureTestingModule({ + providers: [ + ConflictResolutionService, + OperationLogStoreService, + OperationCaptureService, + VectorClockService, + { provide: Store, useValue: storeSpy }, + { provide: OperationApplierService, useValue: applierSpy }, + { provide: ValidateStateService, useValue: validateSpy }, + { provide: OperationLogEffects, useValue: effectsSpy }, + { + provide: SnackService, + useValue: jasmine.createSpyObj('SnackService', ['open']), + }, + { + provide: CLIENT_ID_PROVIDER, + useValue: { + loadClientId: () => Promise.resolve(currentClientId), + getOrGenerateClientId: () => Promise.resolve(currentClientId), + clearCache: () => {}, + }, + }, + { provide: ENTITY_REGISTRY, useValue: entityRegistry }, + ], + }); + + opLogStore = TestBed.inject(OperationLogStoreService); + vectorClock = TestBed.inject(VectorClockService); + resolver = TestBed.inject(ConflictResolutionService); + capture = TestBed.inject(OperationCaptureService); + await opLogStore.init(); + await opLogStore._clearAllDataForTesting(); + }); + + afterEach(async () => { + await opLogStore._clearAllDataForTesting(); + TestBed.resetTestingModule(); + }); + + it('both delivery orders converge on the same title; the synced loser is never rejected; the loser row is recorded-as-seen', async () => { + const clientA = new TestClient(CLIENT_A); + const clientB = new TestClient(CLIENT_B); + const opX = titleUpdateOp(clientA, 'from A', 1_000); + const opY = titleUpdateOp(clientB, 'from B', 2_000); + + // ── Side A: applied+synced X, receives Y (remote wins by timestamp) ── + currentClientId = CLIENT_A; + await installSyncedLocalOp(opX, 'from A'); + + const detectionA = await detect(opY); + expect(detectionA.isSupersededOrDuplicate).toBe(false); + expect(detectionA.conflicts.length).toBe(1); + + const resolutionA = await resolver.autoResolveConflictsLWW( + detectionA.conflicts, + [], + FREEZE_FLAGS, + ); + expect(resolutionA.localWinOpsCreated).toBe(0); + const finalTitleA = getTitle(); + + // The already-synced local side must NOT be rejected (frontier scans skip + // rejected rows — rejecting a synced op would corrupt later detection). + expect((await opLogStore.getOpById(opX.id))?.rejectedAt).toBeUndefined(); + // The winning remote op is durably recorded (ID-dedup shield). + expect(await opLogStore.getOpById(opY.id)).toBeDefined(); + + // ── Side B: applied+synced Y, receives X (local wins → heal op) ── + currentClientId = CLIENT_B; + await installSyncedLocalOp(opY, 'from B'); + + const detectionB = await detect(opX); + expect(detectionB.isSupersededOrDuplicate).toBe(false); + expect(detectionB.conflicts.length).toBe(1); + + const resolutionB = await resolver.autoResolveConflictsLWW( + detectionB.conflicts, + [], + FREEZE_FLAGS, + ); + expect(resolutionB.localWinOpsCreated).toBe(1); + const finalTitleB = getTitle(); + + // CONVERGENCE: both delivery orders end on the deterministic winner. + expect(finalTitleA).toBe('from B'); + expect(finalTitleB).toBe('from B'); + + // The synced local winner is never rejected... + expect((await opLogStore.getOpById(opY.id))?.rejectedAt).toBeUndefined(); + // ...while the remote loser is persisted AND rejected: recorded-as-seen + // (dedup on re-delivery) but excluded from frontier scans and upload. + const loserRow = await opLogStore.getOpById(opX.id); + expect(loserRow).toBeDefined(); + expect(loserRow?.rejectedAt).toBeDefined(); + + // The heal op is a pending upload that dominates BOTH sides of the + // crossing, preserves the winner's timestamp, and carries B's state. + const pendingUploads = await opLogStore.getUnsynced(); + expect(pendingUploads.length).toBe(1); + const healOp = pendingUploads[0].op; + expect(isLwwUpdatePayload(healOp.payload)).toBe(true); + expect( + compareVectorClocks(healOp.vectorClock, { + [CLIENT_A]: opX.vectorClock[CLIENT_A], + [CLIENT_B]: opY.vectorClock[CLIENT_B], + }), + ).toBe(VectorClockComparison.GREATER_THAN); + expect(healOp.timestamp).toBe(2_000); + expect(extractActionPayload(healOp.payload)['title']).toBe('from B'); + + // The heal converges side A too: applying it on top of A's post-resolution + // state is idempotent (same winner state). + localState = updateTaskEntity(initialState, TASK_ID, { title: finalTitleA }); + localState = reducer(localState, convertOpToAction(healOp)); + expect(getTitle()).toBe('from B'); + }); + + it('re-delivery of the resolved loser is skipped as superseded — no second conflict, no second heal', async () => { + const clientA = new TestClient(CLIENT_A); + const clientB = new TestClient(CLIENT_B); + const opX = titleUpdateOp(clientA, 'from A', 1_000); + const opY = titleUpdateOp(clientB, 'from B', 2_000); + + currentClientId = CLIENT_B; + await installSyncedLocalOp(opY, 'from B'); + + const detection = await detect(opX); + await resolver.autoResolveConflictsLWW(detection.conflicts, [], FREEZE_FLAGS); + expect((await opLogStore.getUnsynced()).length).toBe(1); + + // Gap re-download / cursor reset re-delivers the loser: the heal's clock + // now dominates it deterministically. + const redelivery = await detect(opX); + + expect(redelivery.isSupersededOrDuplicate).toBe(true); + expect(redelivery.conflicts).toEqual([]); + expect((await opLogStore.getUnsynced()).length).toBe(1); + expect(getTitle()).toBe('from B'); + }); +}); diff --git a/src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts b/src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts index e83d6af3ae..b4ef7e531a 100644 --- a/src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts +++ b/src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts @@ -234,6 +234,7 @@ describe('round-time conflict convergence integration (#8944)', () => { const detection = await resolver.checkOpForConflicts(remoteTitleOp, { localPendingOpsByEntity: await opLogStore.getUnsyncedByEntity(), appliedFrontierByEntity: new Map(), + retainedOpsByEntity: new Map(), snapshotVectorClock: undefined, snapshotEntityKeys: undefined, hasNoSnapshotClock: true,