mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
A CONCURRENT remote op for a still-existing entity with no pending local ops was applied as-is, so two clients that had each already synced one side of a concurrent pair kept whichever op arrived last and silently diverged forever (arrival-order apply, no conflict, no LWW, no journal). Detection now reconstructs the local side from ALL retained ops still concurrent with the incoming clock (getEntityFrontierWithOps: same scan and rejectedAt filter as the frontier — using only the last op would let a newer disjoint edit mask an older overlapping one) and routes the crossing through the existing autoResolveConflictsLWW pipeline. Both clients therefore resolve the same unordered op pair identically (archive precedence, timestamps, clientId tie), and a local win emits the usual dominating LWW Update op that heals clients whose frontier could not see the crossing. Crossings where apply-both is already lossless and convergent keep today's behavior: identical content, disjoint real fields, noise-only sides, and concurrent task-time deltas. Multi-entity ops and post-compaction crossings also keep the status quo (documented residuals) since a rejection-free resolution cannot compensate atomic siblings. New invariant guard: an already-synced op is never markRejected (frontier scans skip rejected rows; rejecting a synced op would erase it locally while the rest of the fleet keeps it). A no-op for the pending path, whose rejection ids are unsynced by construction. Verified by a mirrored-delivery unit battery (fails 5/13 with the ladder disabled) and an integration spec against the real IndexedDB store with the production freeze flags: both delivery orders converge, the loser row is persisted as recorded-as-seen and rejected, and re-delivery after the heal is skipped without a second resolution.
This commit is contained in:
Johannes Millan
GitHub
parent
afce65dcd7
commit
5a50297215
10 changed files with 1048 additions and 8 deletions
|
|
@ -56,6 +56,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => {
|
|||
interface Ctx {
|
||||
localPendingOpsByEntity: Map<string, Operation[]>;
|
||||
appliedFrontierByEntity: Map<string, VectorClock>;
|
||||
retainedOpsByEntity: Map<string, Operation[]>;
|
||||
snapshotVectorClock: VectorClock | undefined;
|
||||
snapshotEntityKeys: Set<string> | undefined;
|
||||
hasNoSnapshotClock: boolean;
|
||||
|
|
@ -64,6 +65,7 @@ describe('ConflictResolution → ConflictJournal hook (integration)', () => {
|
|||
const ctx = (over: Partial<Ctx> = {}): 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) => ({
|
||||
|
|
|
|||
|
|
@ -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) => ({
|
||||
|
|
|
|||
|
|
@ -6306,18 +6306,21 @@ describe('ConflictResolutionService', () => {
|
|||
const buildCtx = (overrides: {
|
||||
localPendingOpsByEntity?: Map<string, Operation[]>;
|
||||
appliedFrontierByEntity?: Map<string, VectorClock>;
|
||||
retainedOpsByEntity?: Map<string, Operation[]>;
|
||||
snapshotVectorClock?: VectorClock;
|
||||
snapshotEntityKeys?: Set<string>;
|
||||
hasNoSnapshotClock?: boolean;
|
||||
}): {
|
||||
localPendingOpsByEntity: Map<string, Operation[]>;
|
||||
appliedFrontierByEntity: Map<string, VectorClock>;
|
||||
retainedOpsByEntity: Map<string, Operation[]>;
|
||||
snapshotVectorClock: VectorClock | undefined;
|
||||
snapshotEntityKeys: Set<string>;
|
||||
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<string, Operation[]>;
|
||||
appliedFrontierByEntity?: Map<string, VectorClock>;
|
||||
}): {
|
||||
localPendingOpsByEntity: Map<string, Operation[]>;
|
||||
appliedFrontierByEntity: Map<string, VectorClock>;
|
||||
retainedOpsByEntity: Map<string, Operation[]>;
|
||||
snapshotVectorClock: VectorClock | undefined;
|
||||
snapshotEntityKeys: Set<string>;
|
||||
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<string, unknown>;
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<string, Operation[]>;
|
||||
appliedFrontierByEntity: Map<string, VectorClock>;
|
||||
retainedOpsByEntity: Map<string, Operation[]>;
|
||||
snapshotVectorClock: VectorClock | undefined;
|
||||
snapshotEntityKeys: Set<string> | 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<string> | 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<string[]> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<string, Operation[]>(),
|
||||
}));
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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<string, VectorClock>,
|
||||
retainedOpsByEntity: Map<string, Operation[]>,
|
||||
): Promise<ConflictResult> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, VectorClock>;
|
||||
retainedOpsByEntity: Map<string, Operation[]>;
|
||||
}> {
|
||||
const frontier = new Map<string, VectorClock>();
|
||||
const retainedOpsByEntity = new Map<string, Operation[]>();
|
||||
|
||||
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 };
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<RootState, Action>;
|
||||
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<void> => {
|
||||
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<RootState, Action> = (
|
||||
state = initialState,
|
||||
action,
|
||||
) => ({
|
||||
...state,
|
||||
[TASK_FEATURE_NAME]: taskReducer(state[TASK_FEATURE_NAME], action),
|
||||
});
|
||||
reducer = createCombinedTaskSharedMetaReducer(
|
||||
lwwUpdateMetaReducer(rootReducer),
|
||||
) as ActionReducer<RootState, Action>;
|
||||
localState = initialState;
|
||||
currentClientId = CLIENT_A;
|
||||
|
||||
const storeSpy = jasmine.createSpyObj<Store>('Store', ['select']);
|
||||
storeSpy.select.and.callFake((selector: unknown, props?: unknown) => {
|
||||
if (typeof selector !== 'function') {
|
||||
return of(undefined) as ReturnType<Store['select']>;
|
||||
}
|
||||
const selected = (
|
||||
selector as (state: RootState, selectorProps?: unknown) => unknown
|
||||
)(localState, props);
|
||||
return of(selected) as ReturnType<Store['select']>;
|
||||
});
|
||||
|
||||
const applierSpy = jasmine.createSpyObj<OperationApplierService>(
|
||||
'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>(
|
||||
'ValidateStateService',
|
||||
['validateAndRepairCurrentState'],
|
||||
);
|
||||
validateSpy.validateAndRepairCurrentState.and.resolveTo(true);
|
||||
|
||||
const effectsSpy = jasmine.createSpyObj<OperationLogEffects>('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<typeof taskConfig.selectById>;
|
||||
|
||||
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>('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');
|
||||
});
|
||||
});
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue