From 5e754d35525f29be1645c87d122e03228effd37f Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Mon, 13 Jul 2026 21:53:30 +0200 Subject: [PATCH] fix(sync): prevent multi-entity conflict corruption (#8980) * fix(sync): reject disjoint merges for multi-entity ops Scope conflict field extraction and journal titles to the actual entity. Fall back to whole-op LWW whenever either side is multi-entity so sibling updates are never falsely reported as preserved.\n\nFixes #8944. * fix(sync): harden multi-entity conflict resolution --- .../conflict-journal-and-review.md | 28 + packages/sync-core/src/conflict-resolution.ts | 11 +- .../tests/conflict-resolution.spec.ts | 2 +- .../sync/conflict-disjoint-merge.util.spec.ts | 127 +++- .../sync/conflict-disjoint-merge.util.ts | 128 +++- .../sync/conflict-journal-emission.util.ts | 33 +- .../sync/conflict-journal.service.spec.ts | 99 ++++ ...conflict-resolution.disjoint-merge.spec.ts | 560 +++++++++++++++++- .../sync/conflict-resolution.service.spec.ts | 46 +- .../sync/conflict-resolution.service.ts | 357 ++++++++++- ...e-conflict-convergence.integration.spec.ts | 276 +++++++++ src/app/op-log/util/get-op-entity-ids.util.ts | 20 +- 12 files changed, 1608 insertions(+), 79 deletions(-) create mode 100644 src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts diff --git a/docs/sync-and-op-log/conflict-journal-and-review.md b/docs/sync-and-op-log/conflict-journal-and-review.md index e2a751daee..fd1a398234 100644 --- a/docs/sync-and-op-log/conflict-journal-and-review.md +++ b/docs/sync-and-op-log/conflict-journal-and-review.md @@ -83,6 +83,15 @@ delta from two sources in order: 2. the capture-time `entityChanges` computed by `OperationCaptureService` (covers TIME_TRACKING and `syncTimeSpent`). +Extraction is scoped to the entity currently in conflict. `entityId` and +`entityIds` are treated as one deduplicated set, matching server conflict +detection even for inconsistent legacy metadata. For a multi-entity op, only a +matching `entityChanges` entry with `opType: UPDATE`, a plain-object delta, and +no identity (`id`) field is accepted. A multi-entity op without such a safe +target delta is opaque; it must not borrow the primary entity's adapter payload +because doing so could attribute one entity's values to another. Direct-format +legacy bulk payloads are therefore opaque for non-primary entities. + An op with neither is **opaque** (`hasOpaqueChanges`). Opaque ops still represent real state changes, so: @@ -104,6 +113,25 @@ kept by synthesizing a single merged UPDATE op. Eligibility `conflict-resolution.service.ts`): - neither side has a DELETE op, and the plan is not an archive plan; +- neither side contains a multi-entity op. Resolution rejects the original ops, + so merging only the conflicted entity would silently drop the bulk op's + sibling-entity updates. Unsafe partial compensation fails closed before any + op-log mutation, leaving the local operation pending and surfacing a sync + error. Whole-set remote DELETE/archive winners and recreated local archives + retain their existing atomic paths. The one explicitly + decomposable legacy action (`TASK_ROUND_TIME_SPENT`) re-emits its known + per-task time fields from CURRENT state (so a later local edit is not + overwritten). Current round-time capture intentionally emits an empty + `entityChanges` array, so the resolver uses the action's static + `timeSpent`/`timeSpentOnDay` contract only after validating its payload and ID + metadata. This includes a remote-winning conflict target when the remote delta + is safely extractable and disjoint (for example, remote title versus local + rounded time), as well as non-conflicting siblings. Overlapping target + fields remain remote-won only when the remote delta covers the whole coupled + local field set; a partial overlap or opaque remote target delta fails closed. + A sibling missing from current state is not recreated (a later delete owns it). + Arbitrary bulk actions are not split from `entityChanges`: relationship/list + mutations may carry atomic invariants that plain payload shape cannot prove; - neither side has opaque ops (their changes could not be carried into the synthesized delta — merging would silently drop them and the two clients would synthesize DIFFERENT results); diff --git a/packages/sync-core/src/conflict-resolution.ts b/packages/sync-core/src/conflict-resolution.ts index 5a6f8bd9e7..83a0d9f7fd 100644 --- a/packages/sync-core/src/conflict-resolution.ts +++ b/packages/sync-core/src/conflict-resolution.ts @@ -430,11 +430,12 @@ export const partitionLwwResolutions = < partitions.remoteWinsOps.push(...processRemoteWinnerOps(conflict)); for (const op of conflict.remoteOps) { - const ids = op.entityIds?.length - ? op.entityIds - : op.entityId - ? [op.entityId] - : []; + const ids = Array.from( + new Set([ + ...(op.entityId ? [op.entityId] : []), + ...(op.entityIds?.length ? op.entityIds : []), + ]), + ); for (const id of ids) { partitions.remoteWinnerAffectedEntityKeys.add(toEntityKey(op.entityType, id)); } diff --git a/packages/sync-core/tests/conflict-resolution.spec.ts b/packages/sync-core/tests/conflict-resolution.spec.ts index 016810e2a8..5da07d513a 100644 --- a/packages/sync-core/tests/conflict-resolution.spec.ts +++ b/packages/sync-core/tests/conflict-resolution.spec.ts @@ -536,7 +536,7 @@ describe('partitionLwwResolutions', () => { }); expect(result.remoteWinnerAffectedEntityKeys).toEqual( - new Set(['task#task-1', 'task#task-2', 'task#task-3']), + new Set(['task#fallback-task', 'task#task-1', 'task#task-2', 'task#task-3']), ); }); diff --git a/src/app/op-log/sync/conflict-disjoint-merge.util.spec.ts b/src/app/op-log/sync/conflict-disjoint-merge.util.spec.ts index 9a3d673aef..e5fbbc0431 100644 --- a/src/app/op-log/sync/conflict-disjoint-merge.util.spec.ts +++ b/src/app/op-log/sync/conflict-disjoint-merge.util.spec.ts @@ -54,17 +54,102 @@ describe('conflict-disjoint-merge.util', () => { ], }, }); - expect(mergeChangedFields([timeSyncOp], 'task')).toEqual({ + expect(mergeChangedFields([timeSyncOp], 'task', 'task-1')).toEqual({ taskId: 'task-1', date: '2026-07-10', duration: 100, }); }); + + it('does not borrow a direct-format bulk payload from its primary entity', () => { + const bulkOp = op({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + payload: { task: { id: 'task-1', changes: { notes: 'Task 1 notes' } } }, + }); + + expect(mergeChangedFields([bulkOp], 'task', 'task-2')).toEqual({}); + expect(hasOpaqueChanges([bulkOp], 'task', 'task-2')).toBe(true); + }); + + it('requires an adapter payload to positively identify its target entity', () => { + const missingIdOp = op({ + payload: { task: { changes: { notes: 'Unscoped notes' } } }, + }); + + expect(mergeChangedFields([missingIdOp], 'task', 'task-1')).toEqual({}); + expect(hasOpaqueChanges([missingIdOp], 'task', 'task-1')).toBe(true); + }); + + it('treats non-update target entityChanges as opaque', () => { + const bulkOp = op({ + payload: { + actionPayload: { taskId: 'task-1' }, + entityChanges: [ + { + entityType: 'TASK' as EntityType, + entityId: 'task-1', + opType: OpType.Delete, + changes: { title: 'Must not become an update' }, + }, + ], + }, + }); + + expect(mergeChangedFields([bulkOp], 'task', 'task-1')).toEqual({}); + expect(hasOpaqueChanges([bulkOp], 'task', 'task-1')).toBe(true); + }); + + it('treats array and identity-bearing target entityChanges as opaque', () => { + const invalidChanges = [ + ['not', 'a', 'field-map'], + { id: 'task-2', title: 'Must not retarget the update' }, + ]; + + for (const changes of invalidChanges) { + const bulkOp = op({ + payload: { + actionPayload: { taskId: 'task-1' }, + entityChanges: [ + { + entityType: 'TASK' as EntityType, + entityId: 'task-1', + opType: OpType.Update, + changes, + }, + ], + }, + }); + + expect(mergeChangedFields([bulkOp], 'task', 'task-1')).toEqual({}); + expect(hasOpaqueChanges([bulkOp], 'task', 'task-1')).toBe(true); + } + }); + + it('treats a bulk op without a target-specific delta as opaque', () => { + const bulkOp = op({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + payload: { + actionPayload: { taskId: 'task-1' }, + entityChanges: [ + { + entityType: 'TASK' as EntityType, + entityId: 'task-1', + opType: OpType.Update, + changes: { title: 'Task 1' }, + }, + ], + }, + }); + + expect(hasOpaqueChanges([bulkOp], 'task', 'task-2')).toBe(true); + }); }); describe('hasOpaqueChanges', () => { it('is true for a non-adapter payload with no entityChanges (convertToSubTask)', () => { - expect(hasOpaqueChanges([convertToSubTaskOp()], 'task')).toBe(true); + expect(hasOpaqueChanges([convertToSubTaskOp()], 'task', 'task-1')).toBe(true); }); it('is false for adapter-shaped updates and for DELETE ops', () => { @@ -72,12 +157,14 @@ describe('conflict-disjoint-merge.util', () => { hasOpaqueChanges( [op({ payload: { task: { id: 'task-1', title: 'T' } } })], 'task', + 'task-1', ), ).toBe(false); expect( hasOpaqueChanges( [op({ opType: OpType.Delete, payload: { task: { id: 'task-1' } } })], 'task', + 'task-1', ), ).toBe(false); }); @@ -97,9 +184,45 @@ describe('conflict-disjoint-merge.util', () => { op({ payload: { task: { id: 'task-1', notes: 'Remote' } }, clientId: 'B' }), ], payloadKey: 'task', + entityId: 'task-1', }); expect(eligible).toBe(false); }); + + it('refuses inconsistent scalar-plus-array entity metadata', () => { + const mixedMetadataOp = op({ + entityId: 'task-1', + entityIds: ['task-2'], + clientId: 'B', + payload: { + actionPayload: { + task: { id: 'task-1', changes: { notes: 'Task 1 notes' } }, + }, + entityChanges: [ + { + entityType: 'TASK' as EntityType, + entityId: 'task-2', + opType: OpType.Update, + changes: { notes: 'Task 2 notes' }, + }, + ], + }, + }); + + expect( + isDisjointMergeEligible({ + localOps: [ + op({ + entityId: 'task-2', + payload: { task: { id: 'task-2', changes: { title: 'Local' } } }, + }), + ], + remoteOps: [mixedMetadataOp], + payloadKey: 'task', + entityId: 'task-2', + }), + ).toBe(false); + }); }); describe('noiseTiebreakSide', () => { diff --git a/src/app/op-log/sync/conflict-disjoint-merge.util.ts b/src/app/op-log/sync/conflict-disjoint-merge.util.ts index ac2d6003f8..c58bd699f0 100644 --- a/src/app/op-log/sync/conflict-disjoint-merge.util.ts +++ b/src/app/op-log/sync/conflict-disjoint-merge.util.ts @@ -14,8 +14,13 @@ import { OpType } from '../core/operation.types'; import type { Operation } from '../core/operation.types'; -import { extractUpdateChanges, isMultiEntityPayload } from '@sp/sync-core'; +import { + extractEntityFromPayload, + extractUpdateChanges, + isMultiEntityPayload, +} from '@sp/sync-core'; import { ConflictJournalFieldDiff, NOISE_FIELDS } from './conflict-journal.model'; +import { isMultiEntityOperation } from '../util/get-op-entity-ids.util'; /** Identity of one side of the conflict for the deterministic noise tiebreak. */ export interface MergeSideMeta { @@ -26,13 +31,14 @@ export interface MergeSideMeta { } /** - * The changed fields of ONE op, from either of the two delta sources: + * The changed fields of ONE op, scoped to the entity currently in conflict. * - * 1. the adapter-shaped action payload (`{ [payloadKey]: { id, changes } }` - * or a flat entity) via `extractUpdateChanges`; - * 2. falling back to the capture-time `entityChanges` computed by - * `OperationCaptureService` for reducers that don't follow the adapter - * pattern (e.g. TIME_TRACKING, syncTimeSpent). + * Single-entity ops use the adapter-shaped action payload + * (`{ [payloadKey]: { id, changes } }` or a flat entity) first, then fall back + * to capture-time `entityChanges` for reducers that don't follow that pattern + * (e.g. TIME_TRACKING, syncTimeSpent). Multi-entity ops use only the matching + * target-specific `entityChanges` entry; their generic action payload cannot be + * safely attributed to one entity. * * Returns `{}` when neither source has anything — the op's mutation is encoded * in a domain-specific payload shape (e.g. `convertToSubTask`'s @@ -41,25 +47,69 @@ export interface MergeSideMeta { * treat empty-with-payload as unknown, not as "nothing changed" — see * `hasOpaqueChanges`. */ -const extractOpChanges = (op: Operation, payloadKey: string): Record => { - const adapterChanges = extractUpdateChanges(op.payload, payloadKey); - if (Object.keys(adapterChanges).length > 0) { - return adapterChanges; +const asSafeUpdateChanges = (changes: unknown): Record | undefined => { + if (changes === null || typeof changes !== 'object' || Array.isArray(changes)) { + return undefined; } + const record = changes as Record; + return 'id' in record ? undefined : record; +}; + +const extractOpChanges = ( + op: Operation, + payloadKey: string, + entityId: string, +): Record => { + const capturedChanges: Record = {}; if (isMultiEntityPayload(op.payload)) { - const merged: Record = {}; + let hasUnsafeTargetChange = false; for (const change of op.payload.entityChanges) { - if ( - change.entityId === op.entityId && - change.changes && - typeof change.changes === 'object' - ) { - Object.assign(merged, change.changes as Record); + if (change.entityType !== op.entityType || change.entityId !== entityId) { + continue; } + + const safeChanges = + change.opType === OpType.Update ? asSafeUpdateChanges(change.changes) : undefined; + if (!safeChanges) { + hasUnsafeTargetChange = true; + continue; + } + Object.assign(capturedChanges, safeChanges); } - return merged; + + if (hasUnsafeTargetChange) { + return {}; + } + + // A multi-entity op's adapter-shaped action payload is not inherently scoped + // to the entity currently in conflict. Legacy state-diff capture, however, + // recorded one EntityChange per affected entity. Prefer that target-specific + // source exclusively; if it is absent, return {} so the op is treated as + // opaque and falls back to whole-entity LWW instead of borrowing the primary + // entity's fields. + if (isMultiEntityOperation(op)) { + return capturedChanges; + } + } else if (isMultiEntityOperation(op)) { + // Old direct-format bulk payloads describe only their primary entity. They + // cannot be projected onto an arbitrary sibling from entityIds. + return {}; } - return {}; + + const entityPayload = extractEntityFromPayload(op.payload, payloadKey); + const embeddedId = entityPayload?.['id']; + // Adapter entities must positively identify the conflict target. Singleton + // feature state is the sole exception: it uses the '*' sentinel and has no + // embedded id by design. + if (entityId !== '*' && embeddedId !== entityId) { + return capturedChanges; + } + const adapterChanges = extractUpdateChanges(op.payload, payloadKey); + const safeAdapterChanges = asSafeUpdateChanges(adapterChanges); + if (safeAdapterChanges && Object.keys(safeAdapterChanges).length > 0) { + return safeAdapterChanges; + } + return capturedChanges; }; /** @@ -72,21 +122,26 @@ const extractOpChanges = (op: Operation, payloadKey: string): Record => { const merged: Record = {}; for (const op of ops) { if (op.opType === OpType.Delete) { continue; } - Object.assign(merged, extractOpChanges(op, payloadKey)); + Object.assign(merged, extractOpChanges(op, payloadKey, entityId)); } return merged; }; /** True when this non-DELETE op's field-level delta cannot be extracted. */ -export const isOpaqueChangeOp = (op: Operation, payloadKey: string): boolean => +export const isOpaqueChangeOp = ( + op: Operation, + payloadKey: string, + entityId: string, +): boolean => op.opType !== OpType.Delete && - Object.keys(extractOpChanges(op, payloadKey)).length === 0; + Object.keys(extractOpChanges(op, payloadKey, entityId)).length === 0; /** * True when the side contains at least one op whose mutation is real but not @@ -95,8 +150,11 @@ export const isOpaqueChangeOp = (op: Operation, payloadKey: string): boolean => * nor auto-merged (the synthesized entity would silently drop the opaque * mutation and the two clients would diverge). */ -export const hasOpaqueChanges = (ops: Operation[], payloadKey: string): boolean => - ops.some((op) => isOpaqueChangeOp(op, payloadKey)); +export const hasOpaqueChanges = ( + ops: Operation[], + payloadKey: string, + entityId: string, +): boolean => ops.some((op) => isOpaqueChangeOp(op, payloadKey, entityId)); /** The non-NOISE keys of a changed-field map. */ const nonNoiseKeys = (changes: Record): string[] => @@ -126,6 +184,8 @@ export const noiseTiebreakSide = ( * True iff this conflict is safe to resolve by a disjoint-field merge. * * Field-level conditions only (the caller separately excludes archive plans): + * - neither side contains a multi-entity op, because resolving one conflicted + * entity would reject the whole original op and drop its sibling updates; * - neither side has a DELETE op; * - BOTH sides changed at least one real (non-noise) field — if one side only * bumped noise, nothing real is lost by LWW, so leave it to SPAP-13's `noise` @@ -136,8 +196,14 @@ export const isDisjointMergeEligible = (params: { localOps: Operation[]; remoteOps: Operation[]; payloadKey: string; + entityId: string; }): boolean => { - const { localOps, remoteOps, payloadKey } = params; + const { localOps, remoteOps, payloadKey, entityId } = params; + + const hasMultiEntityOp = [...localOps, ...remoteOps].some((op) => + isMultiEntityOperation(op), + ); + if (hasMultiEntityOp) return false; if (localOps.some((op) => op.opType === OpType.Delete)) return false; if (remoteOps.some((op) => op.opType === OpType.Delete)) return false; @@ -145,11 +211,13 @@ export const isDisjointMergeEligible = (params: { // A side with opaque ops has real changes the merge could not carry over — // synthesizing from the extracted fields alone would drop them (and the two // clients would synthesize DIFFERENT entities). Fall back to LWW instead. - if (hasOpaqueChanges(localOps, payloadKey)) return false; - if (hasOpaqueChanges(remoteOps, payloadKey)) return false; + if (hasOpaqueChanges(localOps, payloadKey, entityId)) return false; + if (hasOpaqueChanges(remoteOps, payloadKey, entityId)) return false; - const localNonNoise = nonNoiseKeys(mergeChangedFields(localOps, payloadKey)); - const remoteNonNoise = nonNoiseKeys(mergeChangedFields(remoteOps, payloadKey)); + const localNonNoise = nonNoiseKeys(mergeChangedFields(localOps, payloadKey, entityId)); + const remoteNonNoise = nonNoiseKeys( + mergeChangedFields(remoteOps, payloadKey, entityId), + ); if (localNonNoise.length === 0 || remoteNonNoise.length === 0) return false; const remoteSet = new Set(remoteNonNoise); diff --git a/src/app/op-log/sync/conflict-journal-emission.util.ts b/src/app/op-log/sync/conflict-journal-emission.util.ts index 26174e699d..d6ab3eac8e 100644 --- a/src/app/op-log/sync/conflict-journal-emission.util.ts +++ b/src/app/op-log/sync/conflict-journal-emission.util.ts @@ -27,6 +27,7 @@ import { isOpaqueChangeOp, mergeChangedFields, } from './conflict-disjoint-merge.util'; +import { isMultiEntityOperation } from '../util/get-op-entity-ids.util'; /** Everything the classifier needs about one resolved conflict. */ export interface ConflictJournalClassificationInput { @@ -66,6 +67,7 @@ const extractEntityTitle = ( ops: Operation[], changes: Record, payloadKey: string, + entityId: string, ): string => { const fromChanges = firstString(changes['title'], changes['name']); if (fromChanges) { @@ -75,10 +77,16 @@ const extractEntityTitle = ( const entity = extractEntityFromPayload(op.payload, payloadKey) as | Record | undefined; - const title = firstString(entity?.['title'], entity?.['name']); - if (title) { - return title; + const isMultiEntityOp = isMultiEntityOperation(op); + if (!isMultiEntityOp || entity?.['id'] === entityId) { + const title = firstString(entity?.['title'], entity?.['name']); + if (title) { + return title; + } } + // A multi-entity action's generic payload is not attributable to one target + // entity. Only use its target-scoped `changes`/entity payload above. + if (isMultiEntityOp) continue; // Fallback: some payloads nest the fields directly under the action payload. const action = extractActionPayload(op.payload); const nested = firstString(action?.['title'], action?.['name']); @@ -104,11 +112,12 @@ const buildOpaqueActionDiffs = ( localOps: Operation[], remoteOps: Operation[], payloadKey: string, + entityId: string, pickedSide: 'local' | 'remote', ): ConflictJournalFieldDiff[] => { const byActionType = new Map(); const add = (op: Operation, side: 'local' | 'remote'): void => { - if (!isOpaqueChangeOp(op, payloadKey)) { + if (!isOpaqueChangeOp(op, payloadKey, entityId)) { return; } const diff = byActionType.get(op.actionType) ?? { @@ -162,8 +171,8 @@ export const buildConflictJournalEntry = ( } = input; const payloadKey = resolvePayloadKey(entityType); - const localChanges = mergeChangedFields(localOps, payloadKey); - const remoteChanges = mergeChangedFields(remoteOps, payloadKey); + const localChanges = mergeChangedFields(localOps, payloadKey, entityId); + const remoteChanges = mergeChangedFields(remoteOps, payloadKey, entityId); const localTs = maxTimestamp(localOps); const remoteTs = maxTimestamp(remoteOps); @@ -177,8 +186,8 @@ export const buildConflictJournalEntry = ( const localClientId = localOps[0]?.clientId ?? ''; const remoteClientId = remoteOps[0]?.clientId ?? ''; const mergedTitle = - extractEntityTitle(localOps, localChanges, payloadKey) || - extractEntityTitle(remoteOps, remoteChanges, payloadKey); + extractEntityTitle(localOps, localChanges, payloadKey, entityId) || + extractEntityTitle(remoteOps, remoteChanges, payloadKey, entityId); return { id: uuidv7(), entityType, @@ -216,7 +225,9 @@ export const buildConflictJournalEntry = ( remoteChanged: field in remoteChanges, pickedSide: winner, })); - fieldDiffs.push(...buildOpaqueActionDiffs(localOps, remoteOps, payloadKey, winner)); + fieldDiffs.push( + ...buildOpaqueActionDiffs(localOps, remoteOps, payloadKey, entityId, winner), + ); const winnerOps = winner === 'local' ? localOps : remoteOps; const loserOps = winner === 'local' ? remoteOps : localOps; @@ -227,7 +238,7 @@ export const buildConflictJournalEntry = ( // Opaque loser ops (mutation not readable as fields — e.g. convertToSubTask) // are REAL losses: without this, `loserChanges` is empty and the discarded // structural change would be misclassified as `noise`/`info` and hidden. - const loserHasOpaqueChanges = hasOpaqueChanges(loserOps, payloadKey); + const loserHasOpaqueChanges = hasOpaqueChanges(loserOps, payloadKey, entityId); const isDeleteWin = ARCHIVE_PLAN_REASONS.has(planReason) || @@ -265,11 +276,13 @@ export const buildConflictJournalEntry = ( winnerOps, winner === 'local' ? localChanges : remoteChanges, payloadKey, + entityId, ) || extractEntityTitle( winner === 'local' ? remoteOps : localOps, loserChanges, payloadKey, + entityId, ); return { diff --git a/src/app/op-log/sync/conflict-journal.service.spec.ts b/src/app/op-log/sync/conflict-journal.service.spec.ts index 163cca0c80..4937410aa9 100644 --- a/src/app/op-log/sync/conflict-journal.service.spec.ts +++ b/src/app/op-log/sync/conflict-journal.service.spec.ts @@ -344,6 +344,105 @@ describe('buildConflictJournalEntry (taxonomy)', () => { expect(durationDiff?.localChanged).toBe(true); }); + it('attributes legacy bulk-op field values to the conflicted non-primary entity', () => { + const entry = buildConflictJournalEntry({ + entityType: 'TASK' as EntityType, + entityId: 'task-2', + winner: 'remote', + planReason: 'remote-timestamp-or-tie', + localOps: [ + op({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + payload: { + actionPayload: { + day: '2026-07-10', + taskIds: ['task-1', 'task-2'], + roundTo: 15, + isRoundUp: true, + task: { id: 'task-1', title: 'Wrong primary title' }, + }, + entityChanges: [ + { + entityType: 'TASK' as EntityType, + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK' as EntityType, + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + timestamp: 1000, + }), + ], + remoteOps: [ + op({ + entityId: 'task-2', + payload: { task: { id: 'task-2', changes: { notes: 'Remote' } } }, + timestamp: 2000, + clientId: 'B', + }), + ], + isCorruptionSuspected: false, + resolvePayloadKey, + }); + + const timeSpentDiff = entry.fieldDiffs.find((d) => d.field === 'timeSpent'); + expect(timeSpentDiff?.localVal).toBe(222); + expect(timeSpentDiff?.localChanged).toBe(true); + expect(entry.entityTitle).toBe(''); + }); + + it('keeps a direct-format bulk payload opaque for a non-primary entity', () => { + const entry = buildConflictJournalEntry({ + entityType: 'TASK' as EntityType, + entityId: 'task-2', + winner: 'remote', + planReason: 'remote-timestamp-or-tie', + localOps: [ + op({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + payload: { + task: { + id: 'task-1', + title: 'Wrong primary title', + changes: { notes: 'Task 1 notes' }, + }, + }, + timestamp: 1000, + }), + ], + remoteOps: [ + op({ + entityId: 'task-2', + payload: { task: { id: 'task-2', changes: { title: 'Remote' } } }, + timestamp: 2000, + clientId: 'B', + }), + ], + isCorruptionSuspected: false, + resolvePayloadKey, + }); + + expect(entry.entityTitle).toBe('Remote'); + expect(entry.fieldDiffs.some((diff) => diff.field === 'notes')).toBe(false); + const actionDiff = entry.fieldDiffs.find((diff) => diff.kind === 'action'); + expect(actionDiff?.localChanged).toBe(true); + expect(actionDiff?.localVal).toEqual({ + task: { + id: 'task-1', + title: 'Wrong primary title', + changes: { notes: 'Task 1 notes' }, + }, + }); + }); + it('records per-side presence so winner-only fields are not attributed to the loser', () => { // local (loser) changed only title; remote (winner) changed title + notes. const entry = buildConflictJournalEntry({ 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 026678bcf2..8509a89419 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 @@ -57,21 +57,25 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { ...over, }); - const conflictOf = (localOps: Operation[], remoteOps: Operation[]): EntityConflict => ({ + const conflictOf = ( + localOps: Operation[], + remoteOps: Operation[], + entityId = 'task-1', + ): EntityConflict => ({ entityType: 'TASK', - entityId: 'task-1', + entityId, localOps, remoteOps, suggestedResolution: 'manual', }); - const mergedOpArgs = (): Operation | undefined => + const mergedOpArgs = (entityId = 'task-1'): Operation | undefined => mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls .allArgs() .flatMap(([batches]) => batches) .filter((batch) => batch.source === 'local') .flatMap((batch) => [...batch.ops]) - .find((o) => o.entityId === 'task-1' && o.opType === OpType.Update); + .find((o) => o.entityId === entityId && o.opType === OpType.Update); beforeEach(() => { mockStore = jasmine.createSpyObj('Store', ['select']); @@ -307,6 +311,538 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { expect((await journal.list('unreviewed')).length).toBe(0); }); + it('(a1) fails closed before mutating the op log for a legacy remote bulk op', async () => { + mockStore.select.and.returnValue( + of({ id: 'task-2', title: 'Local title', timeSpent: 0 }), + ); + + const localOp = op({ + id: 'local-task-2', + entityId: 'task-2', + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { task: { id: 'task-2', changes: { title: 'Local title' } } }, + }); + const remoteBulkOp = op({ + id: 'remote-bulk', + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { + actionPayload: { + day: '2026-07-10', + taskIds: ['task-1', 'task-2'], + roundTo: 15, + isRoundUp: true, + }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + + await expectAsync( + service.autoResolveConflictsLWW([conflictOf([localOp], [remoteBulkOp], 'task-2')]), + ).toBeRejectedWithError(/Cannot safely auto-resolve remote multi-entity operation/); + + expect(mergedOpArgs('task-2')).toBeUndefined(); + expect(mockOpLogStore.appendBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); + expect(await journal.list('history')).toEqual([]); + }); + + it('(a1 mirror) refuses disjoint merge for a legacy local bulk op', async () => { + // A later local edit superseded the bulk's captured 111. Reconciliation + // must project the current 333, not resurrect the stale captured value. + let selectCount = 0; + mockStore.select.and.callFake(() => + of( + selectCount++ === 0 + ? { id: 'task-1', timeSpent: 333 } + : { id: 'task-2', timeSpent: 222 }, + ), + ); + mockOpLogStore.getUnsyncedByEntity.and.callFake(async () => { + const writtenTargetReconciliation = + mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local') + .flatMap((batch) => batch.ops) + .find((batchOp) => batchOp.entityId === 'task-2'); + return new Map([ + ['TASK:task-2', writtenTargetReconciliation ? [writtenTargetReconciliation] : []], + ]); + }); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { + day: '2026-07-10', + taskIds: ['task-1', 'task-2'], + roundTo: 15, + isRoundUp: true, + }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { task: { id: 'task-2', changes: { title: 'Remote title' } } }, + }); + + await service.autoResolveConflictsLWW([ + conflictOf([localBulkOp], [remoteOp], 'task-2'), + ]); + + const entries = await journal.list('history'); + expect(entries.length).toBe(1); + expect(entries[0].winner).toBe('remote'); + expect(entries[0].reason).not.toBe('disjoint-merge'); + const timeSpentDiff = entries[0].fieldDiffs.find( + (diff) => diff.field === 'timeSpent', + ); + expect(timeSpentDiff?.localVal).toBe(222); + + const localBatches = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local'); + const siblingReconciliation = localBatches + .flatMap((batch) => batch.ops) + .find((batchOp) => batchOp.entityId === 'task-1'); + const targetReconciliation = localBatches + .flatMap((batch) => batch.ops) + .find((batchOp) => batchOp.entityId === 'task-2'); + expect(siblingReconciliation).toBeDefined(); + expect(siblingReconciliation?.payload).toEqual({ id: 'task-1', timeSpent: 333 }); + expect(targetReconciliation?.payload).toEqual({ id: 'task-2', timeSpent: 222 }); + expect(compareVectorClocks(siblingReconciliation!.vectorClock, { A: 1 })).toBe( + VectorClockComparison.GREATER_THAN, + ); + expect(compareVectorClocks(siblingReconciliation!.vectorClock, { B: 1 })).toBe( + VectorClockComparison.GREATER_THAN, + ); + expect(compareVectorClocks(targetReconciliation!.vectorClock, { B: 1 })).toBe( + VectorClockComparison.GREATER_THAN, + ); + const rejectedIds = mockOpLogStore.markRejected.calls + .allArgs() + .flatMap(([ids]) => ids); + expect(rejectedIds).toContain(localBulkOp.id); + expect(rejectedIds).not.toContain(targetReconciliation!.id); + }); + + it('does not preserve local bulk target fields that overlap a remote winner', async () => { + mockStore.select.and.returnValue(of({ id: 'task-1', timeSpent: 333 })); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { taskIds: ['task-1', 'task-2'] }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { task: { id: 'task-2', changes: { timeSpent: 999 } } }, + }); + + await service.autoResolveConflictsLWW([ + conflictOf([localBulkOp], [remoteOp], 'task-2'), + ]); + + const localOps = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local') + .flatMap((batch) => batch.ops); + expect(localOps.map((batchOp) => batchOp.entityId)).toEqual(['task-1']); + expect(localOps[0].payload).toEqual({ id: 'task-1', timeSpent: 333 }); + }); + + it('fails closed when a remote winner partially overlaps coupled bulk fields', async () => { + const day = '2026-07-10'; + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { taskIds: ['task-1', 'task-2'] }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111, timeSpentOnDay: { [day]: 111 } }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222, timeSpentOnDay: { [day]: 222 } }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { task: { id: 'task-2', changes: { timeSpent: 999 } } }, + }); + + await expectAsync( + service.autoResolveConflictsLWW([conflictOf([localBulkOp], [remoteOp], 'task-2')]), + ).toBeRejectedWithError(/partially overlapping remote winner/); + + expect(mockOpLogStore.appendBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); + expect(await journal.list('history')).toEqual([]); + }); + + it('fails closed when a remote winner is opaque for a local bulk target', async () => { + mockStore.select.and.returnValue(of({ id: 'task-1', timeSpent: 333 })); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { taskIds: ['task-1', 'task-2'] }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { actionPayload: { taskId: 'task-2' } }, + }); + + await expectAsync( + service.autoResolveConflictsLWW([conflictOf([localBulkOp], [remoteOp], 'task-2')]), + ).toBeRejectedWithError(/opaque remote winner/); + + expect(mockOpLogStore.appendBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); + expect(await journal.list('history')).toEqual([]); + }); + + it('does not recreate a bulk sibling deleted by a later local operation', async () => { + let selectCount = 0; + mockStore.select.and.callFake(() => + of(selectCount++ === 0 ? undefined : { id: 'task-2', timeSpent: 222 }), + ); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { taskIds: ['task-1', 'task-2'] }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { task: { id: 'task-2', changes: { title: 'Remote title' } } }, + }); + + await service.autoResolveConflictsLWW([ + conflictOf([localBulkOp], [remoteOp], 'task-2'), + ]); + + const localOps = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local') + .flatMap((batch) => batch.ops); + expect(localOps.map((batchOp) => batchOp.entityId)).toEqual(['task-2']); + expect(localOps[0].payload).toEqual({ id: 'task-2', timeSpent: 222 }); + }); + + it('fails closed for a local bulk action without an explicit decomposition rule', async () => { + const localBulkOp = op({ + id: 'local-opaque-bulk', + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 1000, + payload: { + actionPayload: { taskIds: ['task-1', 'task-2'] }, + entityChanges: [], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 2000, + payload: { task: { id: 'task-2', changes: { title: 'Remote title' } } }, + }); + + await expectAsync( + service.autoResolveConflictsLWW([conflictOf([localBulkOp], [remoteOp], 'task-2')]), + ).toBeRejectedWithError(/Cannot safely auto-resolve local multi-entity operation/); + + expect(mockOpLogStore.appendBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.appendMixedSourceBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); + expect(await journal.list('history')).toEqual([]); + }); + + it('re-emits a decomposable local bulk sibling when the local bulk wins', async () => { + let selectCount = 0; + mockStore.select.and.callFake(() => + of( + selectCount++ === 0 + ? { id: 'task-1', timeSpent: 333 } + : { id: 'task-2', title: 'Base title', timeSpent: 222 }, + ), + ); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 2000, + payload: { + actionPayload: { + day: '2026-07-10', + taskIds: ['task-1', 'task-2'], + roundTo: 15, + isRoundUp: true, + }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteOp = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 1000, + payload: { task: { id: 'task-2', changes: { title: 'Remote title' } } }, + }); + + await service.autoResolveConflictsLWW([ + conflictOf([localBulkOp], [remoteOp], 'task-2'), + ]); + + const localOps = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local') + .flatMap((batch) => batch.ops); + const targetWinner = localOps.find((batchOp) => batchOp.entityId === 'task-2'); + const siblingReconciliation = localOps.find( + (batchOp) => batchOp.entityId === 'task-1', + ); + expect(targetWinner?.payload).toEqual({ + id: 'task-2', + title: 'Base title', + timeSpent: 222, + }); + expect(siblingReconciliation?.payload).toEqual({ id: 'task-1', timeSpent: 333 }); + }); + + it('does not duplicate targets when one local bulk op wins multiple conflicts', async () => { + mockStore.select.and.returnValue( + of({ id: 'selected-task', title: 'Local title', timeSpent: 333 }), + ); + + const localBulkOp = op({ + id: 'local-bulk', + actionType: ActionType.TASK_ROUND_TIME_SPENT, + entityId: 'task-1', + entityIds: ['task-1', 'task-2'], + clientId: 'A', + vectorClock: { A: 1 }, + timestamp: 2000, + payload: { + actionPayload: { + day: '2026-07-10', + taskIds: ['task-1', 'task-2'], + roundTo: 15, + isRoundUp: true, + }, + entityChanges: [ + { + entityType: 'TASK', + entityId: 'task-1', + opType: OpType.Update, + changes: { timeSpent: 111 }, + }, + { + entityType: 'TASK', + entityId: 'task-2', + opType: OpType.Update, + changes: { timeSpent: 222 }, + }, + ], + }, + }); + const remoteTask1 = op({ + id: 'remote-task-1', + entityId: 'task-1', + clientId: 'B', + vectorClock: { B: 1 }, + timestamp: 1000, + payload: { task: { id: 'task-1', changes: { title: 'Remote task 1' } } }, + }); + const remoteTask2 = op({ + id: 'remote-task-2', + entityId: 'task-2', + clientId: 'B', + vectorClock: { B: 2 }, + timestamp: 1000, + payload: { task: { id: 'task-2', changes: { title: 'Remote task 2' } } }, + }); + + await service.autoResolveConflictsLWW([ + conflictOf([localBulkOp], [remoteTask1], 'task-1'), + conflictOf([localBulkOp], [remoteTask2], 'task-2'), + ]); + + const localOps = mockOpLogStore.appendMixedSourceBatchSkipDuplicates.calls + .allArgs() + .flatMap(([batches]) => batches) + .filter((batch) => batch.source === 'local') + .flatMap((batch) => batch.ops); + expect(localOps.filter((batchOp) => batchOp.entityId === 'task-1').length).toBe(1); + expect(localOps.filter((batchOp) => batchOp.entityId === 'task-2').length).toBe(1); + expect(localOps.length).toBe(2); + }); + // ── (a2) merge-only sync counts the synthesized op for re-upload ──────────── it('(a2) counts the synthesized merged op in localWinOpsCreated (drives re-upload)', async () => { mockStore.select.and.returnValue( @@ -594,6 +1130,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { localOps: [localOp], remoteOps: [remoteDelete], payloadKey: 'task', + entityId: 'task-1', }), ).toBe(false); @@ -639,6 +1176,7 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { localOps: [localEdit], remoteOps: [remoteArchive], payloadKey: 'task', + entityId: 'task-1', }), ).toBe(true); @@ -871,10 +1409,20 @@ describe('ConflictResolutionService — SPAP-14 disjoint-field merge', () => { const mA = a1.synthesized!; const mB = b1.synthesized!; expect( - isDisjointMergeEligible({ localOps: [mA], remoteOps: [mB], payloadKey: 'task' }), + isDisjointMergeEligible({ + localOps: [mA], + remoteOps: [mB], + payloadKey: 'task', + entityId: 'task-1', + }), ).toBe(false); expect( - isDisjointMergeEligible({ localOps: [mB], remoteOps: [mA], payloadKey: 'task' }), + isDisjointMergeEligible({ + localOps: [mB], + remoteOps: [mA], + payloadKey: 'task', + entityId: 'task-1', + }), ).toBe(false); }); }); 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 2c19038835..c576f2f4e8 100644 --- a/src/app/op-log/sync/conflict-resolution.service.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.service.spec.ts @@ -1203,7 +1203,7 @@ describe('ConflictResolutionService', () => { expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['remote-mov']); }); - it('should handle BATCH operation conflicts using LWW', async () => { + it('should fail closed for BATCH conflicts that cannot be compensated atomically', async () => { const now = Date.now(); const conflicts: EntityConflict[] = [ { @@ -1229,21 +1229,13 @@ describe('ConflictResolutionService', () => { }, ]; - mockOperationApplier.applyOperations.and.resolveTo({ - appliedOps: conflicts[0].remoteOps, - }); - - await service.autoResolveConflictsLWW(conflicts); - - // Remote BATCH wins (newer timestamp) - expect(mockOpLogStore.appendBatchSkipDuplicates).toHaveBeenCalledWith( - jasmine.arrayContaining([ - jasmine.objectContaining({ id: 'remote-batch', opType: OpType.Batch }), - ]), - 'remote', - jasmine.any(Object), + await expectAsync( + service.autoResolveConflictsLWW(conflicts), + ).toBeRejectedWithError( + /Cannot safely auto-resolve remote multi-entity operation/, ); - expect(mockOpLogStore.markRejected).toHaveBeenCalledWith(['local-batch']); + expect(mockOpLogStore.appendBatchSkipDuplicates).not.toHaveBeenCalled(); + expect(mockOpLogStore.markRejected).not.toHaveBeenCalled(); }); it('should handle singleton entity (GLOBAL_CONFIG) conflicts', async () => { @@ -3619,6 +3611,30 @@ describe('ConflictResolutionService', () => { expect(result.conflict).not.toBeNull(); expect(result.conflict!.entityId).toBe('task-1'); }); + + it('should check both entityId and entityIds when both are present', async () => { + const remoteOp: Operation = { + ...createMockOp('remote-1', 'clientB'), + entityId: 'task-1', + entityIds: ['task-2'], + vectorClock: { clientB: 1 }, + }; + const localOp: Operation = { + ...createMockOp('local-1', 'clientA'), + entityId: 'task-1', + vectorClock: { clientA: 1 }, + }; + const localPendingOpsByEntity = new Map([ + ['TASK:task-1', [localOp]], + ]); + + const result = await service.checkOpForConflicts( + remoteOp, + buildCtx({ localPendingOpsByEntity }), + ); + + expect(result.conflict?.entityId).toBe('task-1'); + }); }); describe('_convertToLWWUpdatesIfNeeded', () => { diff --git a/src/app/op-log/sync/conflict-resolution.service.ts b/src/app/op-log/sync/conflict-resolution.service.ts index af1ef45ebb..d262f56d1f 100644 --- a/src/app/op-log/sync/conflict-resolution.service.ts +++ b/src/app/op-log/sync/conflict-resolution.service.ts @@ -12,6 +12,7 @@ import { isIdenticalConflict as isIdenticalConflictCore, isArrayEntity, isMapEntity, + isMultiEntityPayload, isSingletonEntity, partitionLwwResolutions, planLwwConflictResolutions, @@ -40,7 +41,7 @@ import { HydrationStateService } from '../apply/hydration-state.service'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { OpLog } from '../../core/log'; import { toEntityKey } from '../util/entity-key.util'; -import { getOpEntityIds } from '../util/get-op-entity-ids.util'; +import { getOpEntityIds, isMultiEntityOperation } from '../util/get-op-entity-ids.util'; import { firstValueFrom } from 'rxjs'; import { SnackService } from '../../core/snack/snack.service'; import { BannerService } from '../../core/banner/banner.service'; @@ -68,6 +69,7 @@ import { ConflictJournalService } from './conflict-journal.service'; import { SyncConflictBannerService } from './sync-conflict-banner.service'; import { buildConflictJournalEntry } from './conflict-journal-emission.util'; import { + hasOpaqueChanges, isDisjointMergeEligible, mergeChangedFields, synthesizeMergedChanges, @@ -95,6 +97,7 @@ interface MergedResolution { interface ResolvedConflicts { lwwResolutions: LWWResolution[]; mergedResolutions: MergedResolution[]; + localMultiReconciliationOps: Operation[]; lwwPlans: LwwConflictResolutionPlan[]; } @@ -104,6 +107,51 @@ interface AutoResolveConflictsLwwOptions { remoteApplyLifecycleOwnedByCaller?: boolean; } +// The only legacy bulk operation whose captured per-task deltas are known to be +// independently replayable. Do not generalize this from payload shape alone: +// other multi-entity UPDATE actions encode relationship/list invariants that +// must stay atomic. +const DECOMPOSABLE_MULTI_ACTION_FIELDS = new Map>([ + [ActionType.TASK_ROUND_TIME_SPENT, new Set(['timeSpent', 'timeSpentOnDay'])], +]); + +const isRoundTimePayloadValidForStaticFields = (op: Operation): boolean => { + if (op.actionType !== ActionType.TASK_ROUND_TIME_SPENT) { + return false; + } + const actionPayload = extractActionPayload(op.payload); + const taskIds = actionPayload['taskIds']; + if ( + !Array.isArray(taskIds) || + taskIds.some((id) => typeof id !== 'string') || + typeof actionPayload['day'] !== 'string' || + typeof actionPayload['isRoundUp'] !== 'boolean' + ) { + return false; + } + + const roundTo = actionPayload['roundTo']; + const isKnownRoundOption = + roundTo === undefined || + roundTo === null || + roundTo === '5M' || + roundTo === 'QUARTER' || + roundTo === 'HALF' || + roundTo === 'HOUR' || + // Older payloads represented the interval numerically. + typeof roundTo === 'number'; + if (!isKnownRoundOption) { + return false; + } + + const declaredIds = new Set(taskIds as string[]); + const operationIds = getOpEntityIds(op); + return ( + declaredIds.size === operationIds.length && + operationIds.every((id) => declaredIds.has(id)) + ); +}; + /** * Handles sync conflicts using Last-Write-Wins (LWW) automatic resolution. * @@ -122,7 +170,8 @@ interface AutoResolveConflictsLwwOptions { * * ## Safety Features * - **Duplicate detection**: Skips ops already in the store - * - **Crash safety**: Marks ops as rejected BEFORE applying + * - **Crash safety**: Persists pending replacements before applying and rejects + * originals only after the chosen reducer/archive work succeeds * - **Superseded op rejection**: When remote wins, rejects ALL pending ops for affected entities * (prevents uploading ops with outdated vector clocks) * - **Batch application**: All ops applied together for correct dependency sorting @@ -346,6 +395,7 @@ export class ConflictResolutionService { const { lwwResolutions: resolutions, mergedResolutions, + localMultiReconciliationOps = [], lwwPlans, } = await this._resolveConflictsWithLWW( conflicts, @@ -377,8 +427,10 @@ export class ConflictResolutionService { newLocalWinOps, remoteWinnerAffectedEntityKeys, } = lwwPartitions; + newLocalWinOps.push(...localMultiReconciliationOps); const localOpsToReject = [...lwwPartitions.localOpsToReject]; const localOpsToRejectSet = new Set(localOpsToReject); + const protectedLocalResolutionOpIds = new Set(); let writtenLocalWinOps: Operation[] = []; const writtenMergedOpIds = new Set(); @@ -430,6 +482,7 @@ export class ConflictResolutionService { writtenLocalWinOps = result.written .filter((entry) => entry.source === 'local') .map((entry) => entry.op); + writtenLocalWinOps.forEach((op) => protectedLocalResolutionOpIds.add(op.id)); if (result.skippedCount > 0) { OpLog.verbose( `ConflictResolutionService: Skipped ${result.skippedCount} duplicate mixed-resolution op(s)`, @@ -450,7 +503,10 @@ export class ConflictResolutionService { for (const entityKey of remoteWinnerAffectedEntityKeys) { const pendingOps = pendingByEntity.get(entityKey) || []; for (const op of pendingOps) { - if (!localOpsToRejectSet.has(op.id)) { + if ( + !localOpsToRejectSet.has(op.id) && + !protectedLocalResolutionOpIds.has(op.id) + ) { localOpsToReject.push(op.id); localOpsToRejectSet.add(op.id); OpLog.normal( @@ -927,6 +983,15 @@ export class ConflictResolutionService { toEntityKey: (entityType, entityId) => toEntityKey(entityType as EntityType, entityId), }); + this._assertMultiEntityPlansAreSafe(plans); + + // A rejected local bulk op was already applied optimistically. If the + // remote winner changes only part of one entity, rejecting the whole row + // would strand its other entity/field changes locally with no uploadable op. + // Build safe replacements BEFORE journaling any plan, so a failed safety + // preflight cannot leave a phantom "resolved" journal entry. + const localMultiReconciliationOps = + await this._createLocalMultiReconciliationOps(plans); // SPAP-14 hardening: disjoint-merge is only safe for a SINGLE remote op per // entity per batch. detectConflicts emits one conflict per remote op with no @@ -1011,7 +1076,278 @@ export class ConflictResolutionService { } } - return { lwwResolutions: resolutions, mergedResolutions, lwwPlans }; + return { + lwwResolutions: resolutions, + mergedResolutions, + localMultiReconciliationOps, + lwwPlans, + }; + } + + /** + * Re-emits safely decomposable fields from a local multi-entity op. + * + * The original bulk row is rejected as a unit regardless of which side wins. + * Its disjoint target fields and sibling mutations are still present in the + * local store, so explicitly decomposable fields need new uploadable ops. + * Values are projected from CURRENT entity state, not copied from the old + * captured delta: a later local edit may have superseded the bulk value. + */ + private async _createLocalMultiReconciliationOps( + resolutions: LwwConflictResolutionPlan[], + ): Promise { + const candidates = new Map< + string, + { + entityType: EntityType; + entityId: string; + clocks: VectorClock[]; + fields: Set; + isSafe: boolean; + timestamp: number; + } + >(); + const remoteWholeRemovalKeys = new Set(); + const localWinTargetKeys = new Set(); + const remoteWinnerDiscardedTargetKeys = new Set(); + + for (const resolution of resolutions) { + const conflictTargetKey = toEntityKey( + resolution.conflict.entityType, + resolution.conflict.entityId, + ); + if (resolution.winner === 'local') { + localWinTargetKeys.add(conflictTargetKey); + } + + const remoteRemovalOps = + resolution.winner === 'remote' + ? resolution.conflict.remoteOps.filter( + (op) => + op.opType === OpType.Delete || + op.actionType === ActionType.TASK_SHARED_MOVE_TO_ARCHIVE, + ) + : []; + for (const remoteOp of remoteRemovalOps) { + for (const entityId of getOpEntityIds(remoteOp)) { + remoteWholeRemovalKeys.add(toEntityKey(remoteOp.entityType, entityId)); + } + } + + const conflictPayloadKey = this._resolvePayloadKey(resolution.conflict.entityType); + const remoteWinnerChanges = + resolution.winner === 'remote' && remoteRemovalOps.length === 0 + ? mergeChangedFields( + resolution.conflict.remoteOps, + conflictPayloadKey, + resolution.conflict.entityId, + ) + : {}; + const remoteWinnerIsOpaque = + resolution.winner === 'remote' && + remoteRemovalOps.length === 0 && + hasOpaqueChanges( + resolution.conflict.remoteOps, + conflictPayloadKey, + resolution.conflict.entityId, + ); + + const clocks = [ + ...resolution.conflict.localOps.map((op) => op.vectorClock), + ...resolution.conflict.remoteOps.map((op) => op.vectorClock), + ]; + for (const localOp of resolution.conflict.localOps) { + const allowedFields = DECOMPOSABLE_MULTI_ACTION_FIELDS.get(localOp.actionType); + if (!isMultiEntityOperation(localOp) || !allowedFields) { + continue; + } + for (const entityId of getOpEntityIds(localOp)) { + if ( + resolution.winner === 'local' && + entityId === resolution.conflict.entityId + ) { + // The ordinary local-win full-state op already replaces this target. + continue; + } + const key = toEntityKey(localOp.entityType, entityId); + const existing = candidates.get(key); + const changes = mergeChangedFields( + [localOp], + this._resolvePayloadKey(localOp.entityType), + entityId, + ); + const capturedFields = Object.keys(changes); + const canUseStaticFields = + capturedFields.length === 0 && + isMultiEntityPayload(localOp.payload) && + localOp.payload.entityChanges.length === 0 && + isRoundTimePayloadValidForStaticFields(localOp); + const fields = canUseStaticFields ? [...allowedFields] : capturedFields; + const isRemoteWinTarget = + resolution.winner === 'remote' && entityId === resolution.conflict.entityId; + if (isRemoteWinTarget && remoteWinnerIsOpaque) { + throw new Error( + `ConflictResolutionService: Cannot safely reconcile local bulk fields against ` + + `opaque remote winner for ${resolution.conflict.entityType}:` + + `${resolution.conflict.entityId}`, + ); + } + const remoteOverlappingFields = isRemoteWinTarget + ? fields.filter((field) => field in remoteWinnerChanges) + : []; + if ( + isRemoteWinTarget && + remoteOverlappingFields.length > 0 && + remoteOverlappingFields.length < fields.length + ) { + throw new Error( + `ConflictResolutionService: Cannot safely split coupled local bulk fields against ` + + `partially overlapping remote winner for ${resolution.conflict.entityType}:` + + `${resolution.conflict.entityId}`, + ); + } + if ( + isRemoteWinTarget && + remoteOverlappingFields.length === fields.length && + fields.length > 0 + ) { + // LWW stays authoritative when the remote winner overlaps all + // captured fields. A partial overlap cannot split coupled time fields. + remoteWinnerDiscardedTargetKeys.add(key); + continue; + } + const isSafe = + fields.length > 0 && fields.every((field) => allowedFields.has(field)); + candidates.set(key, { + entityType: localOp.entityType, + entityId, + clocks: [...(existing?.clocks ?? []), ...clocks], + fields: new Set([...(existing?.fields ?? []), ...fields]), + isSafe: (existing?.isSafe ?? true) && isSafe, + timestamp: Math.max(existing?.timestamp ?? 0, localOp.timestamp), + }); + } + } + } + + for (const key of remoteWholeRemovalKeys) { + candidates.delete(key); + } + for (const key of remoteWinnerDiscardedTargetKeys) { + candidates.delete(key); + } + // A local-win conflict target is handled by its ordinary full-state + // replacement. Excluding all such targets globally matters when one bulk + // op participates in more than one conflict: a target skipped in its own + // plan can otherwise be re-added as a "sibling" by another plan. + for (const key of localWinTargetKeys) { + candidates.delete(key); + } + if (candidates.size === 0) { + return []; + } + + const clientId = await this.clientIdProvider.loadClientId(); + if (!clientId) { + throw new Error( + 'ConflictResolutionService: Cannot preserve local bulk siblings - no client ID', + ); + } + + const reconciliationOps: Operation[] = []; + for (const candidate of candidates.values()) { + if (!candidate.isSafe || candidate.fields.size === 0) { + throw new Error( + `ConflictResolutionService: Cannot safely split local multi-entity operation for ` + + `${candidate.entityType}:${candidate.entityId}`, + ); + } + const entityState = await this.getCurrentEntityState( + candidate.entityType, + candidate.entityId, + ); + if (entityState === undefined || entityState === null) { + // A later local delete already superseded the old bulk mutation. Its + // own pending delete op is the authoritative representation; never + // recreate the entity from the stale captured bulk delta. + continue; + } + if (typeof entityState !== 'object' || Array.isArray(entityState)) { + throw new Error( + `ConflictResolutionService: Cannot preserve local bulk sibling - entity state unavailable: ` + + `${candidate.entityType}:${candidate.entityId}`, + ); + } + const stateRecord = entityState as Record; + if ([...candidate.fields].some((field) => !(field in stateRecord))) { + throw new Error( + `ConflictResolutionService: Cannot preserve local bulk sibling - current fields unavailable: ` + + `${candidate.entityType}:${candidate.entityId}`, + ); + } + const currentChanges = Object.fromEntries( + [...candidate.fields].map((field) => [field, stateRecord[field]]), + ); + reconciliationOps.push( + this.createLWWUpdateOp( + candidate.entityType, + candidate.entityId, + currentChanges, + clientId, + this.mergeAndIncrementClocks(candidate.clocks, clientId), + candidate.timestamp, + ), + ); + } + return reconciliationOps; + } + + /** + * Generic multi-entity operations cannot be partially compensated safely. + * Fail before op-log mutation unless the winner removes the whole remote set, + * a local archive is re-created as the same atomic action, or the local legacy + * rounding action has an explicit per-entity reconciliation path above. + */ + private _assertMultiEntityPlansAreSafe( + plans: LwwConflictResolutionPlan[], + ): void { + for (const plan of plans) { + const remoteMultiOps = plan.conflict.remoteOps.filter(isMultiEntityOperation); + const remoteWholeRemovalIsSafe = + plan.winner === 'remote' && + remoteMultiOps.every( + (op) => + op.opType === OpType.Delete || + op.actionType === ActionType.TASK_SHARED_MOVE_TO_ARCHIVE, + ); + if (remoteMultiOps.length > 0 && !remoteWholeRemovalIsSafe) { + throw new Error( + `ConflictResolutionService: Cannot safely auto-resolve remote multi-entity operation ` + + `for ${plan.conflict.entityType}:${plan.conflict.entityId}`, + ); + } + + const localMultiOps = plan.conflict.localOps.filter(isMultiEntityOperation); + const localArchiveIsRecreated = + plan.winner === 'local' && + plan.localWinOperationKind === 'archive-win' && + localMultiOps.every( + (op) => op.actionType === ActionType.TASK_SHARED_MOVE_TO_ARCHIVE, + ); + const localOpsAreDecomposable = localMultiOps.every((op) => + DECOMPOSABLE_MULTI_ACTION_FIELDS.has(op.actionType), + ); + if ( + localMultiOps.length > 0 && + !localArchiveIsRecreated && + !localOpsAreDecomposable + ) { + throw new Error( + `ConflictResolutionService: Cannot safely auto-resolve local multi-entity operation ` + + `for ${plan.conflict.entityType}:${plan.conflict.entityId}`, + ); + } + } } /** @@ -1106,6 +1442,7 @@ export class ConflictResolutionService { localOps: conflict.localOps, remoteOps: conflict.remoteOps, payloadKey, + entityId: conflict.entityId, }) ) { return undefined; @@ -1131,8 +1468,16 @@ export class ConflictResolutionService { return undefined; } - const localChanges = mergeChangedFields(conflict.localOps, payloadKey); - const remoteChanges = mergeChangedFields(conflict.remoteOps, payloadKey); + const localChanges = mergeChangedFields( + conflict.localOps, + payloadKey, + conflict.entityId, + ); + const remoteChanges = mergeChangedFields( + conflict.remoteOps, + payloadKey, + conflict.entityId, + ); const localTs = Math.max(...conflict.localOps.map((op) => op.timestamp)); const remoteTs = Math.max(...conflict.remoteOps.map((op) => op.timestamp)); 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 new file mode 100644 index 0000000000..f49bbb728a --- /dev/null +++ b/src/app/op-log/testing/integration/round-time-conflict-convergence.integration.spec.ts @@ -0,0 +1,276 @@ +import { TestBed } from '@angular/core/testing'; +import { Action, ActionReducer, Store } from '@ngrx/store'; +import { of } from 'rxjs'; +import { ConflictResolutionService } from '../../sync/conflict-resolution.service'; +import { ConflictJournalService } from '../../sync/conflict-journal.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 { Operation } from '../../core/operation.types'; +import { convertOpToAction } from '../../apply/operation-converter.util'; +import { roundTimeSpentForDay } from '../../../features/tasks/store/task.actions'; +import { taskReducer } from '../../../features/tasks/store/task.reducer'; +import { TaskSharedActions } from '../../../root-store/meta/task-shared.actions'; +import { Task } from '../../../features/tasks/task.model'; +import { TASK_FEATURE_NAME } from '../../../features/tasks/store/task.reducer'; +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 { MockSyncServer } from './helpers/mock-sync-server.helper'; +import { resetTestUuidCounter, TestClient } from './helpers/test-client.helper'; + +describe('round-time conflict convergence integration (#8944)', () => { + const DAY = '2026-07-10'; + const TASK_X = 'task-x'; + const TASK_Y = 'task-y'; + const MINUTE = 60_000; + const CLIENT_A = 'round-client-a'; + const CLIENT_B = 'title-client-b'; + + let opLogStore: OperationLogStoreService; + let journal: ConflictJournalService; + let initialState: RootState; + let localState: RootState; + let reducer: ActionReducer; + + const captureOperation = ( + action: PersistentAction, + client: TestClient, + capture: OperationCaptureService, + 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 createReducer = (baseState: RootState): ActionReducer => { + const rootReducer: ActionReducer = ( + state = baseState, + action, + ) => ({ + ...state, + [TASK_FEATURE_NAME]: taskReducer(state[TASK_FEATURE_NAME], action), + }); + return createCombinedTaskSharedMetaReducer( + lwwUpdateMetaReducer(rootReducer), + ) as ActionReducer; + }; + + const getTask = (state: RootState, taskId: string): Task => + state[TASK_FEATURE_NAME].entities[taskId] as Task; + + const taskSyncProjection = (state: RootState, taskId: string): object => { + const task = getTask(state, taskId); + return { + id: task.id, + title: task.title, + timeSpent: task.timeSpent, + timeSpentOnDay: task.timeSpentOnDay, + }; + }; + + beforeEach(async () => { + resetTestUuidCounter(); + + initialState = createStateWithExistingTasks([TASK_X, TASK_Y]); + initialState = updateTaskEntity(initialState, TASK_X, { + title: 'Task X', + timeSpent: 10 * MINUTE, + timeSpentOnDay: { [DAY]: 10 * MINUTE }, + }); + initialState = updateTaskEntity(initialState, TASK_Y, { + title: 'Task Y', + timeSpent: 20 * MINUTE, + timeSpentOnDay: { [DAY]: 20 * MINUTE }, + }); + + reducer = createReducer(initialState); + localState = initialState; + + 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(); + + TestBed.configureTestingModule({ + providers: [ + ConflictResolutionService, + OperationLogStoreService, + OperationCaptureService, + { 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(CLIENT_A), + getOrGenerateClientId: () => Promise.resolve(CLIENT_A), + clearCache: () => {}, + }, + }, + { provide: ENTITY_REGISTRY, useValue: buildEntityRegistry() }, + ], + }); + + opLogStore = TestBed.inject(OperationLogStoreService); + journal = TestBed.inject(ConflictJournalService); + await opLogStore.init(); + await opLogStore._clearAllDataForTesting(); + await journal.clearAll(); + }); + + afterEach(async () => { + await opLogStore._clearAllDataForTesting(); + await journal.clearAll(); + TestBed.resetTestingModule(); + }); + + it('captures, resolves, uploads, replays, and restart-replays without cross-entity corruption', async () => { + const capture = TestBed.inject(OperationCaptureService); + const resolver = TestBed.inject(ConflictResolutionService); + const server = new MockSyncServer(); + const clientA = new TestClient(CLIENT_A); + const clientB = new TestClient(CLIENT_B); + + const roundAction = roundTimeSpentForDay({ + day: DAY, + taskIds: [TASK_X, TASK_Y], + roundTo: 'QUARTER', + isRoundUp: true, + }) as PersistentAction; + localState = reducer(localState, roundAction); + const localBulkOp = captureOperation(roundAction, clientA, capture, 1_000); + + // Current capture intentionally stores action semantics rather than state + // diffs for this reducer-driven bulk action. + expect((localBulkOp.payload as { entityChanges: unknown[] }).entityChanges).toEqual( + [], + ); + await opLogStore.append(localBulkOp, 'local'); + + const remoteTitleAction = TaskSharedActions.updateTask({ + task: { id: TASK_Y, changes: { title: 'Remote title for Y' } }, + }) as PersistentAction; + let remoteState = reducer(initialState, remoteTitleAction); + const remoteTitleOp = captureOperation(remoteTitleAction, clientB, capture, 2_000); + server.uploadOps([remoteTitleOp], CLIENT_B); + + const detection = await resolver.checkOpForConflicts(remoteTitleOp, { + localPendingOpsByEntity: await opLogStore.getUnsyncedByEntity(), + appliedFrontierByEntity: new Map(), + snapshotVectorClock: undefined, + snapshotEntityKeys: undefined, + hasNoSnapshotClock: true, + }); + expect(detection.conflict?.entityId).toBe(TASK_Y); + + const resolution = await resolver.autoResolveConflictsLWW([detection.conflict!]); + expect(resolution.localWinOpsCreated).toBe(2); + + const rejectedBulk = await opLogStore.getOpById(localBulkOp.id); + expect(rejectedBulk?.rejectedAt).toBeDefined(); + + const reconciliationEntries = await opLogStore.getUnsynced(); + const reconciliationOps = reconciliationEntries.map((entry) => entry.op); + expect(reconciliationOps.map((op) => op.entityId).sort()).toEqual([TASK_X, TASK_Y]); + expect(reconciliationEntries.every((entry) => entry.rejectedAt === undefined)).toBe( + true, + ); + + server.uploadOps(reconciliationOps, CLIENT_A); + const downloadedByB = server + .downloadOps(0, CLIENT_B) + .ops.map((entry) => entry.op as Operation); + expect(downloadedByB.length).toBe(2); + for (const op of downloadedByB) { + remoteState = reducer(remoteState, convertOpToAction(op)); + } + + expect(taskSyncProjection(localState, TASK_X)).toEqual( + taskSyncProjection(remoteState, TASK_X), + ); + expect(taskSyncProjection(localState, TASK_Y)).toEqual( + taskSyncProjection(remoteState, TASK_Y), + ); + expect(getTask(localState, TASK_Y).title).toBe('Remote title for Y'); + expect(getTask(localState, TASK_X).timeSpent).toBe(15 * MINUTE); + expect(getTask(localState, TASK_Y).timeSpent).toBe(30 * MINUTE); + + // Simulated restart: rebuild state exclusively from the durable operation + // log, including rejected rows (hydration is deliberately status-blind). + let restartedState = initialState; + const durableEntries = await opLogStore.getOpsAfterSeq(0); + for (const entry of durableEntries) { + restartedState = reducer(restartedState, convertOpToAction(entry.op)); + } + expect(taskSyncProjection(restartedState, TASK_X)).toEqual( + taskSyncProjection(localState, TASK_X), + ); + expect(taskSyncProjection(restartedState, TASK_Y)).toEqual( + taskSyncProjection(localState, TASK_Y), + ); + }); +}); diff --git a/src/app/op-log/util/get-op-entity-ids.util.ts b/src/app/op-log/util/get-op-entity-ids.util.ts index cad7785328..fd33c66ab5 100644 --- a/src/app/op-log/util/get-op-entity-ids.util.ts +++ b/src/app/op-log/util/get-op-entity-ids.util.ts @@ -1,11 +1,23 @@ /** * Normalizes an operation's entity references to a flat id list. * - * Operations carry either `entityIds` (multi-entity) or a single `entityId`. - * Returns the multi list when present, otherwise the single id wrapped in an - * array, otherwise an empty array. + * Operations normally carry either `entityIds` (multi-entity) or a single + * `entityId`, but legacy/malformed operations can contain both. The server's + * conflict detector treats both declarations as authoritative, so the client + * must use the same deduplicated union or it can miss a conflict. */ export const getOpEntityIds = (op: { entityId?: string; entityIds?: string[]; -}): string[] => (op.entityIds?.length ? op.entityIds : op.entityId ? [op.entityId] : []); +}): string[] => + Array.from( + new Set([ + ...(op.entityId ? [op.entityId] : []), + ...(op.entityIds?.length ? op.entityIds : []), + ]), + ); + +export const isMultiEntityOperation = (op: { + entityId?: string; + entityIds?: string[]; +}): boolean => getOpEntityIds(op).length > 1;