diff --git a/.gitignore b/.gitignore index 6aba5beeb3..50c7f99c48 100644 --- a/.gitignore +++ b/.gitignore @@ -45,7 +45,7 @@ android/.idea # misc \\backups /backups -/.angular/cache +/.angular /.sass-cache /connect.lock /coverage diff --git a/packages/super-sync-server/src/sync/conflict.ts b/packages/super-sync-server/src/sync/conflict.ts index 668d382ef8..1aad6b9d99 100644 --- a/packages/super-sync-server/src/sync/conflict.ts +++ b/packages/super-sync-server/src/sync/conflict.ts @@ -246,6 +246,10 @@ export const isSameDuplicateOperation = ( existingOp.opType === op.opType && existingOp.entityType === op.entityType && existingOp.entityId === (op.entityId ?? null) && + // Compare against the same normalization the row was persisted with + // (getStoredEntityIds: single-entity sets collapse to []), so a genuine + // retry matches while a batch op differing only in entityIds does not. + areJsonValuesEqual(existingOp.entityIds, getStoredEntityIds(op)) && payloadsMatch && areJsonValuesEqual(existingOp.vectorClock, storedVectorClock) && existingOp.schemaVersion === op.schemaVersion && diff --git a/packages/super-sync-server/src/sync/sync.types.ts b/packages/super-sync-server/src/sync/sync.types.ts index 0fdc3d5e32..50950b75b6 100644 --- a/packages/super-sync-server/src/sync/sync.types.ts +++ b/packages/super-sync-server/src/sync/sync.types.ts @@ -184,6 +184,7 @@ export interface DuplicateOperationCandidate { opType: string; entityType: string; entityId: string | null; + entityIds: string[]; payload: unknown; vectorClock: unknown; schemaVersion: number; @@ -207,6 +208,7 @@ export const DUPLICATE_OP_SELECT = { opType: true, entityType: true, entityId: true, + entityIds: true, payload: true, vectorClock: true, schemaVersion: true, diff --git a/packages/super-sync-server/tests/conflict.spec.ts b/packages/super-sync-server/tests/conflict.spec.ts index 87f104103b..d0824ddbc0 100644 --- a/packages/super-sync-server/tests/conflict.spec.ts +++ b/packages/super-sync-server/tests/conflict.spec.ts @@ -36,6 +36,7 @@ const duplicateCandidate = ( opType: 'CRT', entityType: 'TASK', entityId: 'task-1', + entityIds: [], payload: { title: 'A' }, vectorClock: { 'client-a': 1 }, schemaVersion: 1, @@ -66,6 +67,57 @@ describe('conflict helpers', () => { ); }); + it('accepts batch retries with identical entityIds', () => { + const incoming = op({ entityIds: ['task-1', 'task-2'] }); + const existing = duplicateCandidate({ entityIds: ['task-1', 'task-2'] }); + + expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(true); + }); + + it('rejects duplicate ids when entityIds differ', () => { + const incoming = op({ entityIds: ['task-1', 'task-3'] }); + const existing = duplicateCandidate({ entityIds: ['task-1', 'task-2'] }); + + expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(false); + }); + + it('rejects duplicate ids when only one side has entityIds', () => { + const incomingWithBatch = op({ entityIds: ['task-1', 'task-2'] }); + expect( + isSameDuplicateOperation(duplicateCandidate(), 1, incomingWithBatch, 60_000), + ).toBe(false); + + const existingWithBatch = duplicateCandidate({ entityIds: ['task-1', 'task-2'] }); + expect(isSameDuplicateOperation(existingWithBatch, 1, op(), 60_000)).toBe(false); + }); + + it('accepts single-entity retries whose entityIds collapse to the scalar entityId', () => { + // getStoredEntityIds persists [] when entityIds is exactly [entityId], so a + // retry that re-sends that redundant array must still match the stored row. + const incoming = op({ entityIds: ['task-1'] }); + + expect(isSameDuplicateOperation(duplicateCandidate(), 1, incoming, 60_000)).toBe( + true, + ); + }); + + it('rejects encrypted retries when entityIds differ', () => { + // With both sides encrypted the payload comparison is skipped, so entityIds + // must independently block a batch-op id collision. + const incoming = op({ + payload: 'BASE64-CIPHERTEXT-A', + isPayloadEncrypted: true, + entityIds: ['task-1', 'task-3'], + }); + const existing = duplicateCandidate({ + payload: 'BASE64-CIPHERTEXT-B', + isPayloadEncrypted: true, + entityIds: ['task-1', 'task-2'], + }); + + expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(false); + }); + it('accepts encrypted retries whose ciphertext differs from the stored payload', () => { // Regression: when encryption is on, encrypt() generates a fresh random IV // per call, so a retry of the same logical op produces different ciphertext. diff --git a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts index 42ad121a8a..e82171f538 100644 --- a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts +++ b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts @@ -441,6 +441,7 @@ describe('Duplicate Operation Pre-check', () => { opType: 'UPD', entityType: 'TASK', entityId: 'task-race', + entityIds: [], payload: { foo: 'bar' }, vectorClock: { 'client-1': 1 }, schemaVersion: 1, @@ -516,6 +517,7 @@ describe('Duplicate Operation Pre-check', () => { opType: 'UPD', entityType: 'TASK', entityId: 'task-other-user', + entityIds: [], payload: { foo: 'bar' }, vectorClock: { 'client-1': 1 }, schemaVersion: 1, diff --git a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts index 3796f83fc9..fd60cfd4a2 100644 --- a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts +++ b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts @@ -93,6 +93,7 @@ const createStoredDuplicateOp = (op: ReturnType) => ({ opType: op.opType, entityType: op.entityType, entityId: op.entityId, + entityIds: [], payload: op.payload, vectorClock: op.vectorClock, schemaVersion: op.schemaVersion, diff --git a/packages/sync-core/src/apply.types.ts b/packages/sync-core/src/apply.types.ts index 8efbbe4024..7c79608c03 100644 --- a/packages/sync-core/src/apply.types.ts +++ b/packages/sync-core/src/apply.types.ts @@ -12,7 +12,11 @@ export interface ApplyOperationsResult = Op /** * If an error occurred, this contains the failed operation and the error. - * Operations after this one in the batch were NOT applied. + * The failed op and the operations after it in the batch did NOT complete + * their archive side effects — but their reducer effects DID commit: the + * bulk dispatch is all-or-nothing and runs before archive handling, so + * archive side effects are the only per-op failure point. Retry paths must + * therefore use `skipReducerDispatch` to avoid double-applying reducers. */ failedOp?: { op: TOperation; @@ -27,4 +31,13 @@ export interface ApplyOperationsOptions { * local operations during hydration. */ isLocalHydration?: boolean; + + /** + * When true, skip the bulk reducer dispatch and run only the post-dispatch + * archive side effects. Used to retry operations whose reducer effects + * already committed in an earlier batch (see `failedOp`): re-dispatching on + * retry would double-apply additive reducers such as time-tracking deltas + * and counter increments. + */ + skipReducerDispatch?: boolean; } diff --git a/packages/sync-core/src/remote-apply.ts b/packages/sync-core/src/remote-apply.ts index a47fe72c97..2546770de6 100644 --- a/packages/sync-core/src/remote-apply.ts +++ b/packages/sync-core/src/remote-apply.ts @@ -74,7 +74,9 @@ const emptyRemoteApplyResult = < * 3. mark successfully applied seqs; * 4. merge applied remote vector clocks; * 5. retain only the newest applied full-state ops when configured; - * 6. mark the failed op and remaining unapplied ops as failed on partial error. + * 6. mark the failed op and remaining ops as failed on partial error — their + * reducer effects committed with the bulk dispatch; only their archive side + * effects are outstanding (see ApplyOperationsResult.failedOp). */ export const applyRemoteOperations = async < TOperation extends Operation = Operation, diff --git a/packages/sync-core/src/replay-coordinator.ts b/packages/sync-core/src/replay-coordinator.ts index 6cc02b825a..567c5374af 100644 --- a/packages/sync-core/src/replay-coordinator.ts +++ b/packages/sync-core/src/replay-coordinator.ts @@ -71,7 +71,9 @@ export const yieldToEventLoop = (): Promise => * package. This coordinator only owns the generic ordering: * * 1. open the remote-apply window; - * 2. dispatch the host's bulk replay action; + * 2. dispatch the host's bulk replay action (skipped when the caller marks + * reducers as already committed via `skipReducerDispatch` — retry paths + * must not double-apply additive reducers); * 3. yield once so host state reducers finish before side effects; * 4. run remote archive side effects after dispatch, when configured; * 5. start post-sync cooldown before closing the remote-apply window; @@ -103,6 +105,7 @@ export const replayOperationBatch = async < } const isLocalHydration = applyOptions.isLocalHydration ?? false; + const skipReducerDispatch = applyOptions.skipReducerDispatch ?? false; if (!isLocalHydration && archiveSideEffects !== undefined && !operationToAction) { throw new Error( @@ -112,8 +115,10 @@ export const replayOperationBatch = async < remoteApplyWindow.startApplyingRemoteOps(); try { - dispatcher.dispatch(createBulkApplyAction(ops)); - await waitForEventLoop(); + if (!skipReducerDispatch) { + dispatcher.dispatch(createBulkApplyAction(ops)); + await waitForEventLoop(); + } if (!isLocalHydration && archiveSideEffects !== undefined && operationToAction) { const archiveResult = await processArchiveSideEffects({ diff --git a/packages/sync-core/src/upload-planning.ts b/packages/sync-core/src/upload-planning.ts index 5be41ce62b..74663d52e3 100644 --- a/packages/sync-core/src/upload-planning.ts +++ b/packages/sync-core/src/upload-planning.ts @@ -11,24 +11,33 @@ export interface PlanRegularOpsAfterFullStateUploadOptions< TEntry extends OperationLogEntry> = OperationLogEntry, > { regularOps: TEntry[]; - lastUploadedFullStateOpId?: string; + lastUploadedFullStateOpSeq?: number; } /** * Splits regular operations around an uploaded full-state snapshot. * - * UUIDv7 operation IDs are time-ordered in Super Productivity. Core only - * depends on lexical ID ordering supplied by the host. Ops before the full-state - * operation are already represented in the snapshot and can be marked synced; - * later ops still need the normal upload path. + * Classification uses the local op-log `seq` (monotonic append order on this + * client), NOT the UUIDv7 op id: lexical id order follows the wall clock, which + * can roll back (e.g. NTP correction, restart), so a post-snapshot op could get + * a smaller id, be treated as "included in snapshot", and silently never reach + * the server. `seq` is exactly creation order, which is what "captured in the + * frozen snapshot payload" means. Ops appended before the full-state op are + * already represented in the snapshot and can be marked synced; later ops still + * need the normal upload path. + * + * When no seq is available (no full-state op uploaded, or its local entry is + * unknown, e.g. the id came from a remote source), everything stays on the + * upload path — uploading an op that is also in the snapshot is safe (server + * dedups by op id), whereas skipping one that is not would lose data. */ export const planRegularOpsAfterFullStateUpload = < TEntry extends OperationLogEntry> = OperationLogEntry, >({ regularOps, - lastUploadedFullStateOpId, + lastUploadedFullStateOpSeq, }: PlanRegularOpsAfterFullStateUploadOptions): RegularOpsAfterFullStateUploadPlan => { - if (!lastUploadedFullStateOpId) { + if (lastUploadedFullStateOpSeq === undefined) { return { opsIncludedInSnapshot: [], opsAfterSnapshot: regularOps, @@ -39,7 +48,7 @@ export const planRegularOpsAfterFullStateUpload = < const opsAfterSnapshot: TEntry[] = []; for (const entry of regularOps) { - if (entry.op.id < lastUploadedFullStateOpId) { + if (entry.seq < lastUploadedFullStateOpSeq) { opsIncludedInSnapshot.push(entry); } else { opsAfterSnapshot.push(entry); diff --git a/packages/sync-core/tests/replay-coordinator.spec.ts b/packages/sync-core/tests/replay-coordinator.spec.ts index faa8cd087e..a94dc95621 100644 --- a/packages/sync-core/tests/replay-coordinator.spec.ts +++ b/packages/sync-core/tests/replay-coordinator.spec.ts @@ -264,6 +264,88 @@ describe('replayOperationBatch', () => { ]); }); + it('runs only archive side effects when skipReducerDispatch is set (retry path)', async () => { + const callOrder: string[] = []; + const ops = [createOperation('op-1'), createOperation('op-2')]; + const dispatcher: ActionDispatchPort = { + dispatch: vi.fn(() => { + callOrder.push('dispatchBulk'); + }), + }; + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async (action) => { + callOrder.push(`archive:${action.opId}`); + }), + }; + + const result = await replayOperationBatch({ + ops, + applyOptions: { skipReducerDispatch: true }, + dispatcher, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + archiveSideEffects, + operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }), + yieldToEventLoop: vi.fn(async () => { + callOrder.push('yield'); + }), + }); + + expect(result).toEqual({ appliedOps: ops }); + expect(dispatcher.dispatch).not.toHaveBeenCalled(); + expect(callOrder).toEqual([ + 'startApplyingRemoteOps', + 'archive:op-1', + 'archive:op-2', + 'yield', + 'startPostSyncCooldown', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('reports archive failures without re-dispatching when skipReducerDispatch is set', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const archiveError = new Error('archive failed again'); + const dispatcher: ActionDispatchPort = { dispatch: vi.fn() }; + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async (action) => { + if (action.opId === 'op-2') { + throw archiveError; + } + }), + }; + + const result = await replayOperationBatch({ + ops: [op1, op2], + applyOptions: { skipReducerDispatch: true }, + dispatcher, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow([]), + deferredLocalActions: createDeferredLocalActions([]), + archiveSideEffects, + operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }), + yieldToEventLoop: vi.fn(async () => undefined), + }); + + expect(dispatcher.dispatch).not.toHaveBeenCalled(); + expect(result).toEqual({ + appliedOps: [op1], + failedOp: { + op: op2, + error: archiveError, + }, + }); + }); + it('fails fast when archive side effects are configured without operation conversion', async () => { const callOrder: string[] = []; const archiveSideEffects: ArchiveSideEffectPort = { diff --git a/packages/sync-core/tests/upload-planning.spec.ts b/packages/sync-core/tests/upload-planning.spec.ts index a9855f3669..9201a59283 100644 --- a/packages/sync-core/tests/upload-planning.spec.ts +++ b/packages/sync-core/tests/upload-planning.spec.ts @@ -32,7 +32,7 @@ describe('planRegularOpsAfterFullStateUpload', () => { expect( planRegularOpsAfterFullStateUpload({ regularOps: entries, - lastUploadedFullStateOpId: undefined, + lastUploadedFullStateOpSeq: undefined, }), ).toEqual({ opsIncludedInSnapshot: [], @@ -40,7 +40,7 @@ describe('planRegularOpsAfterFullStateUpload', () => { }); }); - it('splits regular ops before and after the uploaded full-state op id', () => { + it('splits regular ops before and after the uploaded full-state op seq', () => { const before = createEntry(1, '001'); const same = createEntry(2, '010'); const after = createEntry(3, '011'); @@ -48,13 +48,30 @@ describe('planRegularOpsAfterFullStateUpload', () => { expect( planRegularOpsAfterFullStateUpload({ regularOps: [before, same, after], - lastUploadedFullStateOpId: '010', + lastUploadedFullStateOpSeq: 2, }), ).toEqual({ opsIncludedInSnapshot: [before], opsAfterSnapshot: [same, after], }); }); + + it('uploads an op created after the snapshot even when its id sorts before the full-state op id (clock rollback)', () => { + // Wall-clock rollback: the post-snapshot op got a lexically SMALLER UUIDv7 + // id than the full-state op. Local seq order must win — the op is NOT in + // the frozen snapshot payload and must be uploaded, not marked synced. + const afterSnapshotWithSmallerId = createEntry(5, '001'); + + expect( + planRegularOpsAfterFullStateUpload({ + regularOps: [afterSnapshotWithSmallerId], + lastUploadedFullStateOpSeq: 4, + }), + ).toEqual({ + opsIncludedInSnapshot: [], + opsAfterSnapshot: [afterSnapshotWithSmallerId], + }); + }); }); describe('planUploadLastServerSeqUpdate', () => { diff --git a/src/app/op-log/apply/operation-applier.service.spec.ts b/src/app/op-log/apply/operation-applier.service.spec.ts index b0ea2e7d7e..c923934728 100644 --- a/src/app/op-log/apply/operation-applier.service.spec.ts +++ b/src/app/op-log/apply/operation-applier.service.spec.ts @@ -554,7 +554,10 @@ describe('OperationApplierService', () => { const result = await service.applyOperations(ops); - // Bulk dispatch succeeded (all ops applied to NgRx state) + // Bulk dispatch succeeded: the reducer effects of ALL FIVE ops are + // committed to NgRx state, including the "failed" ones — failedOp only + // means their archive side effects are outstanding. That is why retry + // paths must use skipReducerDispatch (see test below). expect(mockStore.dispatch).toHaveBeenCalledTimes(1); // But archive handling failed on op-3 @@ -566,6 +569,44 @@ describe('OperationApplierService', () => { // Archive handler was called 3 times (op-1, op-2, op-3) expect(mockArchiveOperationHandler.handleOperation).toHaveBeenCalledTimes(3); }); + + it('should not re-run reducers when retrying the failed slice with skipReducerDispatch', async () => { + const ops = [ + createMockOperation('op-1', 'TASK', OpType.Update, { title: 'First' }), + createMockOperation('op-2', 'TASK', OpType.Update, { title: 'Second' }), + createMockOperation('op-3', 'TASK', OpType.Update, { title: 'Third' }), + createMockOperation('op-4', 'TASK', OpType.Update, { title: 'Fourth' }), + createMockOperation('op-5', 'TASK', OpType.Update, { title: 'Fifth' }), + ]; + + let callCount = 0; + mockArchiveOperationHandler.handleOperation.and.callFake(() => { + callCount++; + return callCount === 3 + ? Promise.reject(new Error('Archive write failed on op-3')) + : Promise.resolve(); + }); + + const firstPass = await service.applyOperations(ops); + expect(firstPass.failedOp!.op.id).toBe('op-3'); + expect(mockStore.dispatch).toHaveBeenCalledTimes(1); + + mockStore.dispatch.calls.reset(); + mockArchiveOperationHandler.handleOperation.calls.reset(); + mockArchiveOperationHandler.handleOperation.and.returnValue(Promise.resolve()); + + // Retry the failed slice the way retryFailedRemoteOps does: archive only. + const retry = await service.applyOperations(ops.slice(2), { + skipReducerDispatch: true, + }); + + // No reducer re-dispatch — additive reducers (e.g. syncTimeSpent, + // increaseSimpleCounterCounterToday) cannot double-apply on retry. + expect(mockStore.dispatch).not.toHaveBeenCalled(); + expect(mockArchiveOperationHandler.handleOperation).toHaveBeenCalledTimes(3); + expect(retry.appliedOps).toEqual(ops.slice(2)); + expect(retry.failedOp).toBeUndefined(); + }); }); describe('effects isolation (key architectural benefit)', () => { diff --git a/src/app/op-log/apply/operation-applier.service.ts b/src/app/op-log/apply/operation-applier.service.ts index 0d7df161c4..4f0dc8b4c1 100644 --- a/src/app/op-log/apply/operation-applier.service.ts +++ b/src/app/op-log/apply/operation-applier.service.ts @@ -106,7 +106,10 @@ export class OperationApplierService implements OperationApplyPort { const result = await replayOperationBatch({ ops, - applyOptions: { isLocalHydration }, + applyOptions: { + isLocalHydration, + skipReducerDispatch: options.skipReducerDispatch, + }, dispatcher: this.store, createBulkApplyAction: (operations) => bulkApplyOperations({ operations, localClientId }), diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts index 0f0adab79d..21d9b4d1ea 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.spec.ts @@ -7,6 +7,8 @@ import { bufferDeferredAction, getDeferredActions, clearDeferredActions, + DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD, + DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP, } from './operation-capture.meta-reducer'; import { OperationCaptureService } from './operation-capture.service'; import { Action } from '@ngrx/store'; @@ -269,6 +271,86 @@ describe('operationCaptureMetaReducer', () => { expect(getDeferredActions()).toEqual([]); }); }); + + describe('buffer limits', () => { + // devError shows a native alert + confirm (and throws if confirm returns + // true), so force confirm to return false. src/test.ts installs a + // PERMANENT global confirm spy (jasmine.createSpy, never auto-restored), + // so reset its accumulated calls per test and restore the global + // returnValue(true) default afterwards. + const spyNativeDialogs = (): jasmine.Spy => { + if (!jasmine.isSpy(window.alert)) { + spyOn(window, 'alert'); + } + const confirmSpy = jasmine.isSpy(window.confirm) + ? (window.confirm as jasmine.Spy) + : spyOn(window, 'confirm'); + confirmSpy.calls.reset(); + confirmSpy.and.returnValue(false); + return confirmSpy; + }; + + afterEach(() => { + if (jasmine.isSpy(window.confirm)) { + (window.confirm as jasmine.Spy).and.returnValue(true); + } + }); + + const createManyActions = (count: number): PersistentAction[] => + Array.from({ length: count }, (_, i) => + createMockAction({ type: `[Test] Action ${i}` }), + ); + + it('should preserve ALL actions in order past the reload-warning threshold', () => { + spyNativeDialogs(); + const actions = createManyActions(DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD + 50); + + actions.forEach((a) => bufferDeferredAction(a)); + + // Nothing may be dropped: each buffered action's state change was + // already accepted into NgRx — dropping = permanent unsyncable divergence. + expect(getDeferredActions()).toEqual(actions); + }); + + it('should fire devError at the reload-warning threshold without dropping', () => { + const confirmSpy = spyNativeDialogs(); + const actions = createManyActions(DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD); + const reloadWarningCalls = (): unknown[][] => + confirmSpy.calls + .allArgs() + .filter((args) => /consider reloading/.test(String(args[0]))); + + actions.slice(0, -1).forEach((a) => bufferDeferredAction(a)); + expect(reloadWarningCalls().length).toBe(0); + + bufferDeferredAction(actions[actions.length - 1]); + expect(reloadWarningCalls().length).toBe(1); + + expect(getDeferredActions()).toEqual(actions); + }); + + it('should drop the oldest action only past the pathological hard cap', () => { + const confirmSpy = spyNativeDialogs(); + const actions = createManyActions(DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP); + const dropErrorCalls = (): unknown[][] => + confirmSpy.calls + .allArgs() + .filter((args) => /pathological cap/.test(String(args[0]))); + + actions.forEach((a) => bufferDeferredAction(a)); + expect(dropErrorCalls().length).toBe(0); + + const extraAction = createMockAction({ type: '[Test] Extra Action' }); + bufferDeferredAction(extraAction); + expect(dropErrorCalls().length).toBe(1); + + const buffered = getDeferredActions(); + expect(buffered.length).toBe(DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP); + // Oldest action was dropped, remaining order intact + expect(buffered[0]).toBe(actions[1]); + expect(buffered[buffered.length - 1]).toBe(extraAction); + }); + }); }); describe('sync buffering (user interaction during sync)', () => { diff --git a/src/app/op-log/capture/operation-capture.meta-reducer.ts b/src/app/op-log/capture/operation-capture.meta-reducer.ts index 54cfaf89c0..31a49d714c 100644 --- a/src/app/op-log/capture/operation-capture.meta-reducer.ts +++ b/src/app/op-log/capture/operation-capture.meta-reducer.ts @@ -133,31 +133,48 @@ export const getIsApplyingRemoteOps = (): boolean => { }; /** - * Maximum number of deferred actions before warning. + * Soft-warning threshold for the deferred actions buffer. * If exceeded, sync may be stuck or taking too long. */ -const MAX_DEFERRED_ACTIONS_WARNING = 10; +const DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD = 10; /** - * Hard limit for deferred actions buffer. - * If reached, oldest actions are dropped to prevent unbounded memory growth. + * Reload-warning threshold: a devError advises the user to reload, but + * NOTHING is dropped. Every buffered action's state change was already + * accepted into NgRx; dropping it would mean no operation ever represents + * it — a permanent, unsyncable local divergence. + * Exported for tests. */ -const MAX_DEFERRED_ACTIONS_HARD_LIMIT = 100; +export const DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD = 100; + +/** + * Pathological hard cap for the deferred actions buffer — purely a memory + * backstop against a runaway dispatch loop. ~100 buffered actions are + * plausibly reachable by a real user during a stuck/very long remote-apply + * window, so we must never drop there (see reload-warning threshold above, + * which tells the user to reload long before this cap can be hit through + * real interaction). Only past this cap is drop-oldest the lesser evil vs + * unbounded memory growth. + * Exported for tests. + */ +export const DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP = 5000; /** * Buffers an action for processing after sync completes. * Called by the meta-reducer when a persistent action arrives during sync. */ export const bufferDeferredAction = (action: PersistentAction): void => { - // Hard limit: drop oldest action if buffer is full (sync stuck scenario). - // NOTE: The shifted action remains in deferredActionSet (WeakSet has no delete-by-value). - // The effect filters it via isDeferredAction(), and getDeferredActions() won't return it, - // so it is silently lost. This is acceptable: the hard limit is itself an error condition - // (sync stuck), and dropping the oldest action is the lesser evil vs unbounded growth. - if (deferredActions.length >= MAX_DEFERRED_ACTIONS_HARD_LIMIT) { + // Pathological cap only: dropping an accepted persistent action means its + // state mutation survives in NgRx while no operation will ever represent + // it → permanent, unsyncable divergence. So this drop exists solely as a + // memory backstop against a runaway dispatch loop, never as flow control. + // NOTE: The shifted action remains in deferredActionSet (WeakSet has no + // delete-by-value). The effect filters it via isDeferredAction(), and + // getDeferredActions() won't return it, so it is silently lost. + if (deferredActions.length >= DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP) { devError( - `[operationCaptureMetaReducer] Deferred actions buffer exceeded ${MAX_DEFERRED_ACTIONS_HARD_LIMIT} items. ` + - `Dropping oldest action. Sync may be stuck - consider reloading the app.`, + `[operationCaptureMetaReducer] Deferred actions buffer exceeded pathological cap of ${DEFERRED_ACTIONS_PATHOLOGICAL_HARD_CAP} items. ` + + `Dropping oldest action - this local change will NOT sync. Likely a runaway dispatch loop; reload the app.`, ); deferredActions.shift(); } @@ -165,8 +182,12 @@ export const bufferDeferredAction = (action: PersistentAction): void => { deferredActions.push(action); deferredActionSet.add(action); - // Soft warning at 10 items - if (deferredActions.length > MAX_DEFERRED_ACTIONS_WARNING) { + if (deferredActions.length >= DEFERRED_ACTIONS_RELOAD_WARNING_THRESHOLD) { + devError( + `[operationCaptureMetaReducer] Deferred actions buffer has ${deferredActions.length} items. ` + + `Sync may be stuck - consider reloading the app. Nothing is dropped; actions remain buffered.`, + ); + } else if (deferredActions.length > DEFERRED_ACTIONS_SOFT_WARNING_THRESHOLD) { devError( `[operationCaptureMetaReducer] Deferred actions buffer has ${deferredActions.length} items - sync may be stuck or taking too long`, ); diff --git a/src/app/op-log/core/types/apply.types.ts b/src/app/op-log/core/types/apply.types.ts index 3e80976225..7c2e72f24b 100644 --- a/src/app/op-log/core/types/apply.types.ts +++ b/src/app/op-log/core/types/apply.types.ts @@ -7,6 +7,11 @@ import type { Operation } from '../operation.types'; export interface ApplyOperationsResult { /** Operations that were successfully applied. */ appliedOps: Operation[]; + /** + * First op whose archive side effect threw. Its reducer effect (and that of + * every op after it) DID commit — the bulk dispatch is all-or-nothing and + * precedes archive handling — so retry paths must pass `skipReducerDispatch`. + */ failedOp?: { op: Operation; error: Error; @@ -27,4 +32,13 @@ export interface ApplyOperationsOptions { * their vector clocks. */ skipDeferredLocalActions?: boolean; + + /** + * When true, skip the bulk reducer dispatch and run only the archive side + * effects. Used to retry ops marked `failed`, whose reducer effects already + * committed (see `ApplyOperationsResult.failedOp`): re-dispatching would + * double-apply additive reducers such as syncTimeSpent or + * increaseSimpleCounterCounterToday. + */ + skipReducerDispatch?: boolean; } diff --git a/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts b/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts index bec40e8500..aea87db639 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.retry.integration.spec.ts @@ -132,6 +132,12 @@ describe('OperationLogHydratorService retryFailedRemoteOps (integration, real st expect(applier.applyOperations).toHaveBeenCalledTimes(1); const passed = applier.applyOperations.calls.argsFor(0)[0] as Operation[]; expect(passed.length).toBe(3); + // Archive side effects only: the failed ops' reducers committed in the + // batch that marked them failed, so a reducer re-dispatch would + // double-apply additive reducers. + expect(applier.applyOperations.calls.argsFor(0)[1]).toEqual({ + skipReducerDispatch: true, + }); }); it('on partial failure marks the failed op and every op after it as still-failing', async () => { diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts index 3ba60e317a..6c77d85278 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts @@ -410,6 +410,50 @@ describe('OperationLogHydratorService', () => { expect(mockHydrationStateService.endApplyingRemoteOps).toHaveBeenCalled(); }); + it('should apply a failed tail op exactly once across hydration replay + retry (one boot)', async () => { + // Regression: a remote op marked 'failed' (archive side effect threw + // after its reducer committed) with seq > lastAppliedOpSeq used to get + // its reducer applied TWICE per boot — once by the status-blind tail + // replay and once more by retryFailedRemoteOps re-dispatching. The + // retry must complete it with archive side effects only. + const snapshot = createMockSnapshot({ lastAppliedOpSeq: 5 }); + const failedOp = createMockOperation('op-failed'); + const failedTailEntry: OperationLogEntry = { + seq: 6, + op: failedOp, + appliedAt: Date.now(), + source: 'remote', + applicationStatus: 'failed', + retryCount: 1, + }; + mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); + mockOpLogStore.getOpsAfterSeq.and.returnValue(Promise.resolve([failedTailEntry])); + mockOpLogStore.getFailedRemoteOps.and.returnValue( + Promise.resolve([failedTailEntry]), + ); + mockOperationApplierService.applyOperations.and.callFake((ops: Operation[]) => + Promise.resolve({ appliedOps: ops }), + ); + + await service.hydrateStore(); + + // Reducer application happens exactly once: the tail replay bulk dispatch. + const bulkDispatches = mockStore.dispatch.calls + .allArgs() + .map((args) => args[0] as unknown as { type: string; operations?: Operation[] }) + .filter((action) => action.type === bulkApplyHydrationOperations.type); + expect(bulkDispatches.length).toBe(1); + expect(bulkDispatches[0].operations!.map((o) => o.id)).toEqual(['op-failed']); + + // The retry completes the op WITHOUT re-dispatching its reducer. + expect(mockOperationApplierService.applyOperations).toHaveBeenCalledTimes(1); + const [retriedOps, retryOptions] = + mockOperationApplierService.applyOperations.calls.argsFor(0); + expect(retriedOps.map((o: Operation) => o.id)).toEqual(['op-failed']); + expect(retryOptions).toEqual({ skipReducerDispatch: true }); + expect(mockOpLogStore.markApplied).toHaveBeenCalledWith([6]); + }); + it('should request ops after snapshot sequence', async () => { const snapshot = createMockSnapshot({ lastAppliedOpSeq: 42 }); mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); @@ -1364,6 +1408,26 @@ describe('OperationLogHydratorService', () => { expect(passedOps.map((o: Operation) => o.id)).toEqual(['op-a', 'op-b', 'op-c']); expect(mockOpLogStore.markApplied).toHaveBeenCalledWith([40, 41, 42]); expect(mockOpLogStore.markFailed).not.toHaveBeenCalled(); + // Applied-op clocks are merged on completion (parity with the primary + // remote-apply path, which only merges the clocks of ops it marked applied). + expect(mockOpLogStore.mergeRemoteOpClocks).toHaveBeenCalledWith(passedOps); + }); + + it('should retry archive side effects only — never re-dispatch reducers', async () => { + // Failed ops had their reducers committed by the bulk dispatch of the + // batch that marked them failed; re-dispatching on retry double-applies + // additive reducers (syncTimeSpent, increaseSimpleCounterCounterToday). + mockOpLogStore.getFailedRemoteOps.and.returnValue( + Promise.resolve([failedEntry(40, 'op-a')]), + ); + mockOperationApplierService.applyOperations.and.callFake((ops: Operation[]) => + Promise.resolve({ appliedOps: ops }), + ); + + await service.retryFailedRemoteOps(); + + const options = mockOperationApplierService.applyOperations.calls.argsFor(0)[1]; + expect(options).toEqual({ skipReducerDispatch: true }); }); it('should apply failed ops in ascending seq order regardless of store order', async () => { diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.ts b/src/app/op-log/persistence/operation-log-hydrator.service.ts index c2b092fac4..e35345841e 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.ts @@ -208,6 +208,21 @@ export class OperationLogHydratorService { ); // 4. Replay tail operations (A.7.13: with operation migration) + // + // Replay is deliberately status-blind (getOpsAfterSeq has no status + // filter) — every entry's reducer effect belongs in state exactly once: + // - applied ops: their effect is state history by definition. + // - failed ops (remote, archive side effect threw): their reducers DID + // commit before the failure (bulk dispatch precedes archive handling), + // so replay restores that effect; retryFailedRemoteOps() below then + // re-runs ONLY the outstanding archive side effects. + // - rejected ops: every rejection path appends its compensation AFTER + // them in seq order, so replay converges to post-resolution runtime + // state — server-rejected local ops are followed by merged ops + // (SupersededOperationResolver) or keep their effect (permanent + // rejections never revert state), and LWW-losing remote ops are + // followed by the local-win op that overwrites them + // (ConflictResolutionService). const tailOps = await this.opLogStore.getOpsAfterSeq(snapshot.lastAppliedOpSeq); if (tailOps.length > 0) { @@ -293,6 +308,8 @@ export class OperationLogHydratorService { ); // No snapshot means we might be in a fresh install state or post-migration-check with no legacy data. // We must replay ALL operations from the beginning of the log. + // Status-blind on purpose — see the replay-policy note in the snapshot + // branch above. const allOps = await this.opLogStore.getOpsAfterSeq(0); if (allOps.length === 0) { @@ -557,8 +574,10 @@ export class OperationLogHydratorService { * Called after hydration to give failed ops another chance to apply now that * more state might be available (e.g., dependencies resolved by sync). * - * Failed ops are ops that previously failed during conflict resolution - * but may succeed now that more state has been loaded. + * Failed ops are ops whose archive side effect threw after their reducers + * committed, so the retry runs archive side effects ONLY + * (`skipReducerDispatch`) — hydration replay / the snapshot already carry + * their reducer effects. */ async retryFailedRemoteOps(): Promise { const failedOps = await this.opLogStore.getFailedRemoteOps(); @@ -583,7 +602,17 @@ export class OperationLogHydratorService { const opsToApply = orderedFailedOps.map((e) => e.op); const opIdToSeq = new Map(orderedFailedOps.map((e) => [e.op.id, e.seq])); - const result = await this.operationApplierService.applyOperations(opsToApply); + // `failed` can only be set AFTER a bulk dispatch committed (archive side + // effects run after the dispatch and are the only per-op failure point), + // so every failed op's reducer effect is already in state — via the + // snapshot when its seq <= lastAppliedOpSeq, via the status-blind tail + // replay above otherwise. Skip the reducer dispatch and re-run only the + // outstanding archive side effects: re-dispatching would double-apply + // additive reducers (syncTimeSpent, increaseSimpleCounterCounterToday) + // on every retry attempt. + const result = await this.operationApplierService.applyOperations(opsToApply, { + skipReducerDispatch: true, + }); // Mark successfully applied ops. const appliedSeqs = result.appliedOps @@ -591,6 +620,10 @@ export class OperationLogHydratorService { .filter((seq): seq is number => seq !== undefined); if (appliedSeqs.length > 0) { await this.opLogStore.markApplied(appliedSeqs); + // Parity with the primary remote-apply path: applyRemoteOperations only + // merges clocks for ops it marked applied, so a failed op's clock is + // merged here, when its application completes. + await this.opLogStore.mergeRemoteOpClocks(result.appliedOps); OpLog.normal( `OperationLogHydratorService: Successfully retried ${appliedSeqs.length} failed ops`, ); diff --git a/src/app/op-log/sync/operation-log-download.service.ts b/src/app/op-log/sync/operation-log-download.service.ts index c551689a84..daf055fb21 100644 --- a/src/app/op-log/sync/operation-log-download.service.ts +++ b/src/app/op-log/sync/operation-log-download.service.ts @@ -72,9 +72,18 @@ export class OperationLogDownloadService implements OnDestroy { this.clockDriftRetryServerTimestamp = null; } + /** + * @param options.includeOwnAndAppliedOps - Raw-rebuild mode for USE_REMOTE: + * skips the appliedOpIds filter AND downloads ops authored by this + * client (no excludeClient param). Required to reconstruct the full + * server history — the normal filters would drop everything the local + * store already knows, leaving a destructive replace with nothing to + * replay. Callers are expected to clear the local op store before + * appending the result. + */ async downloadRemoteOps( syncProvider: OperationSyncCapable, - options?: { forceFromSeq0?: boolean }, + options?: { forceFromSeq0?: boolean; includeOwnAndAppliedOps?: boolean }, ): Promise { if (!syncProvider) { OpLog.warn( @@ -88,7 +97,7 @@ export class OperationLogDownloadService implements OnDestroy { private async _downloadRemoteOpsViaApi( syncProvider: OperationSyncCapable, - options?: { forceFromSeq0?: boolean }, + options?: { forceFromSeq0?: boolean; includeOwnAndAppliedOps?: boolean }, ): Promise { const forceFromSeq0 = options?.forceFromSeq0 ?? false; OpLog.normal( @@ -117,8 +126,14 @@ export class OperationLogDownloadService implements OnDestroy { await this.lockService.request(LOCK_NAMES.DOWNLOAD, async () => { const lastServerSeq = forceFromSeq0 ? 0 : await syncProvider.getLastServerSeq(); - const appliedOpIds = await this.opLogStore.getAppliedOpIds(); - const clientId = await this.clientIdProvider.loadClientId(); + // Raw-rebuild mode: an empty applied set disables the duplicate filter + // below; a missing clientId makes the server include this client's own ops. + const appliedOpIds = options?.includeOwnAndAppliedOps + ? new Set() + : await this.opLogStore.getAppliedOpIds(); + const clientId = options?.includeOwnAndAppliedOps + ? null + : await this.clientIdProvider.loadClientId(); OpLog.verbose( `OperationLogDownloadService: [DEBUG] Starting download. ` + `lastServerSeq=${lastServerSeq}, appliedOpIds.size=${appliedOpIds.size}, clientId=${clientId}`, diff --git a/src/app/op-log/sync/operation-log-sync.service.spec.ts b/src/app/op-log/sync/operation-log-sync.service.spec.ts index dde580c6bc..f0cc485793 100644 --- a/src/app/op-log/sync/operation-log-sync.service.spec.ts +++ b/src/app/op-log/sync/operation-log-sync.service.spec.ts @@ -107,6 +107,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }); rejectedOpsHandlerServiceSpy = jasmine.createSpyObj('RejectedOpsHandlerService', [ @@ -327,6 +328,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }); const mockProvider = { @@ -439,6 +441,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }; }); const setLastServerSeqSpy = jasmine @@ -690,6 +693,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }); // handleRejectedOps returns 3 merged ops created @@ -877,6 +881,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }); const mockProvider = { @@ -892,6 +897,55 @@ describe('OperationLogSyncService', () => { } }); + it('should NOT advance lastServerSeq when processing blocked at an incompatible op', async () => { + const remoteOp: Operation = { + id: 'op-future', + clientId: 'client-B', + actionType: 'test' as ActionType, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + payload: {}, + vectorClock: { clientB: 1 }, + timestamp: Date.now(), + schemaVersion: 99, + }; + + downloadServiceSpy.downloadRemoteOps.and.returnValue( + Promise.resolve({ + newOps: [remoteOp], + hasMore: false, + latestSeq: 5, + latestServerSeq: 5, + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + }), + ); + + remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({ + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, + }); + + const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); + const mockProvider = { + isReady: () => Promise.resolve(true), + setLastServerSeq: setLastServerSeqSpy, + } as any; + + const result = await service.downloadRemoteOps(mockProvider); + + // Cursor stays behind the blocked op so it is re-downloaded and retried + // after an app update instead of skipped forever. + expect(result.kind).toBe('ops_processed'); + expect(setLastServerSeqSpy).not.toHaveBeenCalled(); + }); + it('should return localWinOpsCreated: 0 and newOpsCount: 0 on server migration', async () => { downloadServiceSpy.downloadRemoteOps.and.returnValue( Promise.resolve({ @@ -955,6 +1009,7 @@ describe('OperationLogSyncService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }; }); @@ -2318,29 +2373,44 @@ describe('OperationLogSyncService', () => { describe('forceDownloadRemoteState', () => { let downloadServiceSpy: jasmine.SpyObj; + const makeRemoteOp = (id: string = 'op1'): Operation => ({ + id, + actionType: 'ACTION' as ActionType, + opType: 'UPDATE' as OpType, + entityType: 'TASK', + entityId: 'task1', + payload: {}, + clientId: 'remote', + vectorClock: { remote: 1 }, + timestamp: Date.now(), + schemaVersion: 1, + }); + beforeEach(() => { downloadServiceSpy = TestBed.inject( OperationLogDownloadService, ) as jasmine.SpyObj; - opLogStoreSpy.clearUnsyncedOps = jasmine - .createSpy('clearUnsyncedOps') + opLogStoreSpy.clearAllOperations = jasmine + .createSpy('clearAllOperations') .and.resolveTo(); + opLogStoreSpy.saveStateCache = jasmine.createSpy('saveStateCache').and.resolveTo(); }); - it('should clear unsynced ops before downloading', async () => { + it('should download BEFORE any destructive local mutation', async () => { const callOrder: string[] = []; - opLogStoreSpy.clearUnsyncedOps.and.callFake(async () => { - callOrder.push('clearUnsyncedOps'); + opLogStoreSpy.clearAllOperations.and.callFake(async () => { + callOrder.push('clearAllOperations'); }); downloadServiceSpy.downloadRemoteOps.and.callFake(async () => { callOrder.push('downloadRemoteOps'); return { - newOps: [], + newOps: [makeRemoteOp()], needsFullStateUpload: false, success: true, providerMode: 'superSyncOps', failedFileCount: 0, + latestServerSeq: 1, }; }); @@ -2351,24 +2421,25 @@ describe('OperationLogSyncService', () => { await service.forceDownloadRemoteState(mockProvider); - expect(callOrder[0]).toBe('clearUnsyncedOps'); + expect(callOrder).toEqual(['downloadRemoteOps', 'clearAllOperations']); }); - it('should capture a safety backup BEFORE clearing unsynced ops (#8107)', async () => { + it('should capture a safety backup BEFORE clearing local data (#8107)', async () => { const callOrder: string[] = []; backupServiceSpy.captureImportBackup.and.callFake(async () => { callOrder.push('captureImportBackup'); return 1; }); - opLogStoreSpy.clearUnsyncedOps.and.callFake(async () => { - callOrder.push('clearUnsyncedOps'); + opLogStoreSpy.clearAllOperations.and.callFake(async () => { + callOrder.push('clearAllOperations'); }); downloadServiceSpy.downloadRemoteOps.and.resolveTo({ - newOps: [], + newOps: [makeRemoteOp()], needsFullStateUpload: false, success: true, providerMode: 'superSyncOps', failedFileCount: 0, + latestServerSeq: 1, }); const mockProvider = { supportsOperationSync: true, @@ -2377,7 +2448,7 @@ describe('OperationLogSyncService', () => { await service.forceDownloadRemoteState(mockProvider); - expect(callOrder).toEqual(['captureImportBackup', 'clearUnsyncedOps']); + expect(callOrder).toEqual(['captureImportBackup', 'clearAllOperations']); }); it('should ABORT without wiping local data if the safety backup fails (#8107)', async () => { @@ -2389,18 +2460,18 @@ describe('OperationLogSyncService', () => { await expectAsync(service.forceDownloadRemoteState(mockProvider)).toBeRejected(); - expect(opLogStoreSpy.clearUnsyncedOps).not.toHaveBeenCalled(); - expect(opLogStoreSpy.clearFullStateOps).not.toHaveBeenCalled(); + expect(opLogStoreSpy.clearAllOperations).not.toHaveBeenCalled(); expect(downloadServiceSpy.downloadRemoteOps).not.toHaveBeenCalled(); }); it('should offer to restore the previous data after replacing (#8107)', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ - newOps: [], + newOps: [makeRemoteOp()], needsFullStateUpload: false, success: true, providerMode: 'superSyncOps', failedFileCount: 0, + latestServerSeq: 1, }); const mockProvider = { supportsOperationSync: true, @@ -2419,11 +2490,12 @@ describe('OperationLogSyncService', () => { it('should reset lastServerSeq to 0', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ - newOps: [], + newOps: [makeRemoteOp()], needsFullStateUpload: false, success: true, providerMode: 'superSyncOps', failedFileCount: 0, + latestServerSeq: 1, }); const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); @@ -2438,13 +2510,14 @@ describe('OperationLogSyncService', () => { expect(setLastServerSeqSpy).toHaveBeenCalledWith(0); }); - it('should download ops with forceFromSeq0 option', async () => { + it('should download raw history: forceFromSeq0 AND includeOwnAndAppliedOps', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ - newOps: [], + newOps: [makeRemoteOp()], needsFullStateUpload: false, success: true, providerMode: 'superSyncOps', failedFileCount: 0, + latestServerSeq: 1, }); const mockProvider = { @@ -2454,13 +2527,13 @@ describe('OperationLogSyncService', () => { await service.forceDownloadRemoteState(mockProvider); - expect(downloadServiceSpy.downloadRemoteOps).toHaveBeenCalledWith( - mockProvider, - jasmine.objectContaining({ forceFromSeq0: true }), - ); + expect(downloadServiceSpy.downloadRemoteOps).toHaveBeenCalledWith(mockProvider, { + forceFromSeq0: true, + includeOwnAndAppliedOps: true, + }); }); - it('should throw when force download fails', async () => { + it('should throw when force download fails and leave local data untouched', async () => { downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [], needsFullStateUpload: false, @@ -2468,6 +2541,30 @@ describe('OperationLogSyncService', () => { failedFileCount: 1, }); + const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: setLastServerSeqSpy, + } as any; + + await expectAsync( + service.forceDownloadRemoteState(mockProvider), + ).toBeRejectedWithError(/Download failed/); + expect(remoteOpsProcessingServiceSpy.processRemoteOps).not.toHaveBeenCalled(); + expect(opLogStoreSpy.clearAllOperations).not.toHaveBeenCalled(); + expect(setLastServerSeqSpy).not.toHaveBeenCalled(); + }); + + it('should refuse to rebuild from ops with a newer schema version BEFORE destroying anything', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [{ ...makeRemoteOp('op-future'), schemaVersion: 99 }], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); + const mockProvider = { supportsOperationSync: true, setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), @@ -2475,10 +2572,41 @@ describe('OperationLogSyncService', () => { await expectAsync( service.forceDownloadRemoteState(mockProvider), - ).toBeRejectedWithError(/Download failed/); + ).toBeRejectedWithError(/newer schema version/); + expect(opLogStoreSpy.clearAllOperations).not.toHaveBeenCalled(); expect(remoteOpsProcessingServiceSpy.processRemoteOps).not.toHaveBeenCalled(); }); + it('should NOT advance the cursor past 0 when replay blocks on a migration failure', async () => { + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 50, + }); + remoteOpsProcessingServiceSpy.processRemoteOps.and.resolveTo({ + localWinOpsCreated: 0, + allOpsFilteredBySyncImport: false, + filteredOpCount: 0, + isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, + }); + + const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); + const mockProvider = { + supportsOperationSync: true, + setLastServerSeq: setLastServerSeqSpy, + } as any; + + await expectAsync( + service.forceDownloadRemoteState(mockProvider), + ).toBeRejectedWithError(/USE_REMOTE incomplete/); + expect(setLastServerSeqSpy).toHaveBeenCalledWith(0); + expect(setLastServerSeqSpy).not.toHaveBeenCalledWith(50); + }); + it('should process downloaded ops without confirmation', async () => { const mockOps: Operation[] = [ { @@ -2554,7 +2682,9 @@ describe('OperationLogSyncService', () => { expect(setLastServerSeqSpy).toHaveBeenCalledWith(50); }); - it('should handle empty remote state gracefully', async () => { + it('should REJECT an empty remote instead of silently succeeding', async () => { + // An empty remote is not a state to adopt: succeeding here used to wipe + // the local op-log bookkeeping while leaving live state unchanged. downloadServiceSpy.downloadRemoteOps.and.resolveTo({ newOps: [], needsFullStateUpload: false, @@ -2563,13 +2693,18 @@ describe('OperationLogSyncService', () => { failedFileCount: 0, }); + const setLastServerSeqSpy = jasmine.createSpy('setLastServerSeq').and.resolveTo(); const mockProvider = { supportsOperationSync: true, - setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), + setLastServerSeq: setLastServerSeqSpy, } as any; - await expectAsync(service.forceDownloadRemoteState(mockProvider)).toBeResolved(); + await expectAsync( + service.forceDownloadRemoteState(mockProvider), + ).toBeRejectedWithError(/no data to rebuild from/); expect(remoteOpsProcessingServiceSpy.processRemoteOps).not.toHaveBeenCalled(); + expect(opLogStoreSpy.clearAllOperations).not.toHaveBeenCalled(); + expect(setLastServerSeqSpy).not.toHaveBeenCalled(); }); it('should hydrate from snapshotState when present (file-based sync)', async () => { @@ -2618,12 +2753,21 @@ describe('OperationLogSyncService', () => { expect(setLastServerSeqSpy).toHaveBeenCalledWith(1); }); - it('should propagate errors from clearUnsyncedOps', async () => { + it('should propagate errors from clearAllOperations', async () => { const error = new Error('Failed to clear ops'); - opLogStoreSpy.clearUnsyncedOps.and.rejectWith(error); + opLogStoreSpy.clearAllOperations.and.rejectWith(error); + downloadServiceSpy.downloadRemoteOps.and.resolveTo({ + newOps: [makeRemoteOp()], + needsFullStateUpload: false, + success: true, + providerMode: 'superSyncOps', + failedFileCount: 0, + latestServerSeq: 1, + }); const mockProvider = { supportsOperationSync: true, + setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(), } as any; await expectAsync(service.forceDownloadRemoteState(mockProvider)).toBeRejectedWith( @@ -3568,6 +3712,68 @@ describe('OperationLogSyncService', () => { expect(result.kind).toBe('cancelled'); }); + it('should show conflict dialog for local work accepted in the SAME upload round (pre-upload snapshot race)', async () => { + const piggybackedSyncImport: Operation = { + id: 'import-1', + clientId: 'client-B', + actionType: ActionType.LOAD_ALL_DATA, + opType: OpType.SyncImport, + entityType: 'ALL', + payload: { task: { ids: ['remote-task'] } }, + vectorClock: { clientB: 5 }, + timestamp: Date.now(), + schemaVersion: 1, + }; + + uploadServiceSpy.uploadPendingOps.and.resolveTo({ + uploadedCount: 1, + piggybackedOps: [piggybackedSyncImport], + rejectedCount: 0, + rejectedOps: [], + }); + + const pendingEntry: OperationLogEntry = { + seq: 1, + op: { + id: 'local-op-1', + clientId: 'client-A', + actionType: 'test' as ActionType, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + payload: { title: 'Local Title' }, + vectorClock: { clientA: 1 }, + timestamp: Date.now(), + schemaVersion: 1, + }, + appliedAt: Date.now(), + source: 'local', + }; + // The op is pending BEFORE the upload but marked synced DURING it (server + // accepted it in the same round that piggybacked the import) — a live + // post-upload read no longer sees it. + let getUnsyncedCalls = 0; + opLogStoreSpy.getUnsynced.and.callFake(async () => { + getUnsyncedCalls++; + return getUnsyncedCalls === 1 ? [pendingEntry] : []; + }); + + syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL'); + + const mockProvider = { + isReady: () => Promise.resolve(true), + } as any; + + const result = await service.uploadPendingOps(mockProvider); + + // Without the pre-upload snapshot, the gate would read the (now empty) + // live pending set and silently accept the import over the local edit. + expect(syncImportConflictDialogServiceSpy.showConflictDialog).toHaveBeenCalledWith( + jasmine.objectContaining({ scenario: 'INCOMING_IMPORT' }), + ); + expect(result.kind).toBe('cancelled'); + }); + it('should process piggybacked SYNC_IMPORT silently when client only has already-synced meaningful data', async () => { const piggybackedSyncImport: Operation = { id: 'import-1', diff --git a/src/app/op-log/sync/operation-log-sync.service.ts b/src/app/op-log/sync/operation-log-sync.service.ts index 61801c3758..7fd655e5f8 100644 --- a/src/app/op-log/sync/operation-log-sync.service.ts +++ b/src/app/op-log/sync/operation-log-sync.service.ts @@ -10,7 +10,6 @@ import { import { FILE_BASED_SYNC_CONSTANTS } from '../sync-providers/file-based/file-based-sync.types'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { BackupService } from '../backup/backup.service'; -import { FULL_STATE_OP_TYPES } from '../core/operation.types'; import { OpLog } from '../../core/log'; import { OperationSyncCapable, @@ -39,6 +38,7 @@ import { SyncImportConflictResolution, } from './dialog-sync-import-conflict/dialog-sync-import-conflict.component'; import { SyncImportConflictGateService } from './sync-import-conflict-gate.service'; +import { CURRENT_SCHEMA_VERSION } from '../persistence/schema-migration.service'; import { SyncProviderManager } from '../sync-providers/provider-manager.service'; import { getDefaultMainModelData } from '../model/model-config'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; @@ -209,6 +209,14 @@ export class OperationLogSyncService { return { kind: 'blocked_fresh_client' }; } + // Capture pending ops BEFORE the upload round (the flush at the top of this + // method already drained in-flight writes). The piggyback conflict gate below + // must judge "would this import discard local work?" against this snapshot: + // ops accepted by the server during the upload are marked synced per chunk, + // so a post-upload getUnsynced() read no longer sees local work that a + // piggybacked import is about to replace. + const pendingOpsAtUploadStart = await this.opLogStore.getUnsynced(); + // SERVER MIGRATION CHECK: Passed as callback to execute INSIDE the upload lock. // This prevents race conditions where multiple tabs could both detect migration // and create duplicate SYNC_IMPORT operations. @@ -248,7 +256,10 @@ export class OperationLogSyncService { const piggybackedConflict = await this.syncImportConflictGateService.checkIncomingFullStateConflict( result.piggybackedOps, - { isNeverSynced: isNeverSyncedAtSyncStart }, + { + isNeverSynced: isNeverSyncedAtSyncStart, + preCapturedPendingOps: pendingOpsAtUploadStart, + }, ); if (piggybackedConflict.fullStateOp) { const { fullStateOp, pendingOps, dialogData } = piggybackedConflict; @@ -313,7 +324,12 @@ export class OperationLogSyncService { // so that a crash — or a cancelled/USE_REMOTE/USE_LOCAL SYNC_IMPORT dialog, all of // which return early ABOVE without reaching here — cannot advance the seq past ops // that were never stored. Mirrors the download path's invariant. - if (result.lastServerSeqToPersist !== undefined) { + // A version/migration block keeps the cursor behind the blocked op so it is + // re-downloaded and retried after an app update instead of skipped forever. + if ( + result.lastServerSeqToPersist !== undefined && + !processResult.blockedByIncompatibleOp + ) { await syncProvider.setLastServerSeq(result.lastServerSeqToPersist); } } @@ -915,7 +931,9 @@ export class OperationLogSyncService { // This ensures localStorage and IndexedDB stay in sync. If we crash before this point, // lastServerSeq won't be updated, and the client will re-download the ops on next sync. // This is the correct behavior - better to re-download than to skip ops. - if (result.latestServerSeq !== undefined) { + // A version/migration block keeps the cursor behind the blocked op so it is + // re-downloaded and retried after an app update instead of skipped forever. + if (result.latestServerSeq !== undefined && !processResult.blockedByIncompatibleOp) { await syncProvider.setLastServerSeq(result.latestServerSeq); } @@ -1059,8 +1077,9 @@ export class OperationLogSyncService { ); // Persist the cursor only AFTER the ops are applied, matching the normal - // incremental path's crash-safety ordering. - if (result.latestServerSeq !== undefined) { + // incremental path's crash-safety ordering. A version/migration block keeps + // the cursor behind the blocked op (retried after an app update). + if (result.latestServerSeq !== undefined && !processResult.blockedByIncompatibleOp) { await syncProvider.setLastServerSeq(result.latestServerSeq); } @@ -1120,17 +1139,21 @@ export class OperationLogSyncService { * Force download all remote state, replacing local data. * This is used when user explicitly chooses "USE_REMOTE" in conflict resolution. * - * Clears all local unsynced operations and downloads from seq 0 to get - * the complete remote state including any SYNC_IMPORT. + * Download-first rebuild: the COMPLETE server history (including this + * client's own ops and ops already known locally) is downloaded and + * validated BEFORE any local mutation. Only then is the local op-log + * replaced wholesale with the server history and replayed from a defaults + * baseline. Aborts without touching local state on download failure, an + * empty remote, or a remote containing newer-schema ops. * - * IMPORTANT: This also resets the vector clock to the remote snapshot's clock + * IMPORTANT: This also resets the vector clock to the remote's rebuilt clock * to ensure rejected local ops don't pollute the causal history. * * @param syncProvider - The sync provider to download from */ async forceDownloadRemoteState(syncProvider: OperationSyncCapable): Promise { OpLog.warn( - 'OperationLogSyncService: Force downloading remote state - clearing local import and unsynced ops.', + 'OperationLogSyncService: Force downloading remote state for a full rebuild.', ); // Safety net (#8107): snapshot current local state before we destroy it, so an @@ -1151,25 +1174,21 @@ export class OperationLogSyncService { ); } - // IMPORTANT: Clear local full-state ops (SYNC_IMPORT, BACKUP_IMPORT, REPAIR) - // This is critical - if the user chose USE_REMOTE, we must not filter incoming - // ops against the local import that we're discarding. - const clearedFullStateOps = await this.opLogStore.clearFullStateOps(); - if (clearedFullStateOps > 0) { - OpLog.normal( - `OperationLogSyncService: Cleared ${clearedFullStateOps} local full-state op(s).`, - ); - } - - // Clear all unsynced local ops - we're replacing them with remote state - await this.opLogStore.clearUnsyncedOps(); - - // Reset lastServerSeq to 0 so we download everything - await syncProvider.setLastServerSeq(0); - - // Download all remote ops from the beginning + // ───────────────────────────────────────────────────────────────────────── + // PHASE 1 — download and validate. Nothing local is mutated until the + // complete server history is in memory: a network failure here must leave + // local state untouched (previously the op-log was cleared and the cursor + // reset BEFORE the download, so a failed download stranded the client with + // destroyed local bookkeeping). + // + // Raw-rebuild mode: includes ops authored by this client and ops already + // known locally. The normal download filters drop both, which made the old + // USE_REMOTE unable to rebuild a server history this client had (fully or + // partially) produced itself. + // ───────────────────────────────────────────────────────────────────────── const result = await this.downloadService.downloadRemoteOps(syncProvider, { forceFromSeq0: true, + includeOwnAndAppliedOps: true, }); if (!result.success) { @@ -1179,20 +1198,60 @@ export class OperationLogSyncService { ); } - // Reset the vector clock to the remote snapshot's clock. - // This removes entries from rejected local ops that would otherwise - // pollute the causal history and cause incorrect conflict detection. - if (result.snapshotVectorClock) { - await this.opLogStore.setVectorClock(result.snapshotVectorClock); - OpLog.normal( - 'OperationLogSyncService: Reset vector clock to remote snapshot clock.', + const hasSnapshotState = + result.providerMode === 'fileSnapshotOps' && !!result.snapshotState; + if (!hasSnapshotState && result.newOps.length === 0) { + // An empty remote is not a state to adopt. Silently succeeding here used + // to leave live NgRx state untouched while the op-log bookkeeping was + // already wiped — state and log permanently disagreeing. + throw new Error('USE_REMOTE aborted: remote returned no data to rebuild from.'); + } + + // Refuse BEFORE destroying anything if the remote history contains ops this + // client cannot interpret (newer schema) — replay would block midway and + // leave a partial rebuild. Migration exceptions can still surface during + // replay; those are handled after processRemoteOps below. + const tooNewOp = result.newOps.find( + (op) => (op.schemaVersion ?? 1) > CURRENT_SCHEMA_VERSION, + ); + if (tooNewOp) { + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.VERSION_TOO_OLD, + actionStr: T.PS.UPDATE_APP, + actionFn: () => window.open('https://super-productivity.com/download', '_blank'), + }); + throw new Error( + 'USE_REMOTE aborted: remote history contains ops from a newer schema version — update the app first.', ); } + // ───────────────────────────────────────────────────────────────────────── + // PHASE 2 — destructive replace (local mutations only, no network left). + // The local op-log becomes exactly the server history: clearing ALL ops + // (not just unsynced/full-state ones) lets the raw replay below re-append + // and re-apply ops this client already knew. The old keep-synced-ops + // variant made the duplicate filter skip them, so a "rebuild" replayed only + // an unseen suffix onto a defaults reset. + // ───────────────────────────────────────────────────────────────────────── + await this.opLogStore.clearAllOperations(); + await syncProvider.setLastServerSeq(0); + + // Reset the vector clock to the remote's causal knowledge (snapshot clock + // merged with every downloaded op clock). This also drops entries from + // rejected local ops that would otherwise pollute conflict detection. + let rebuiltClock: VectorClock = { ...(result.snapshotVectorClock ?? {}) }; + for (const opClock of result.allOpClocks ?? []) { + rebuiltClock = mergeVectorClocks(rebuiltClock, opClock); + } + await this.opLogStore.setVectorClock(rebuiltClock); + OpLog.normal('OperationLogSyncService: Reset vector clock to rebuilt remote clock.'); + // FILE-BASED SYNC: Handle snapshot state from force download. // When downloading from seq 0 on file-based providers, we may receive a // snapshotState instead of incremental ops. This happens when the remote // has a SYNC_IMPORT (full state snapshot) with empty recentOps. + // hydrateFromRemoteSync persists its own state cache + vector clock. if (result.providerMode === 'fileSnapshotOps' && result.snapshotState) { OpLog.normal( 'OperationLogSyncService: Force download received snapshotState. Hydrating...', @@ -1235,48 +1294,59 @@ export class OperationLogSyncService { return; } - if (result.newOps.length > 0) { - // Check if there's a full-state op in the batch (SYNC_IMPORT would replace state) - const hasFullStateOp = result.newOps.some((op) => - FULL_STATE_OP_TYPES.has(op.opType), + // Crash safety: persist a defaults baseline so an interrupted replay + // hydrates as defaults + appended prefix on next boot (a consistent prefix + // of server state, completed by the next sync from cursor 0) instead of + // layering the new history onto the stale pre-replace snapshot. + await this.opLogStore.saveStateCache({ + state: getDefaultMainModelData(), + lastAppliedOpSeq: 0, + vectorClock: rebuiltClock, + compactedAt: Date.now(), + snapshotEntityKeys: [], + }); + + // Reset live state to defaults, then replay the COMPLETE server history on + // top. A full-state op in the history replaces state again by its own + // semantics; a purely incremental history rebuilds from this baseline. + const defaultData = getDefaultMainModelData(); + this.store.dispatch( + loadAllData({ + appDataComplete: defaultData as Parameters< + typeof loadAllData + >[0]['appDataComplete'], + }), + ); + // Brief yield to let NgRx process the state reset + await new Promise((resolve) => setTimeout(resolve, 0)); + + // Process all remote ops (no confirmation needed - user already chose USE_REMOTE). + // Skip conflict detection because the NgRx store was just reset to empty state, + // which causes all entities to appear missing and CONCURRENT ops to be discarded. + // Validation failure is surfaced via the session-validation latch. (#7330) + const processResult = await this.remoteOpsProcessingService.processRemoteOps( + result.newOps, + { skipConflictDetection: true }, + ); + + if (processResult.blockedByIncompatibleOp) { + // Version blocks were pre-checked above; only a migration exception lands + // here. The rebuild is partial: keep the cursor at 0 so the next sync + // retries the remainder, and surface the failure — the Undo snack still + // offers the pre-replace backup. + this._showRestorePreviousDataSnack(backupSavedAt); + throw new Error( + 'USE_REMOTE incomplete: an op failed schema migration during replay.', ); + } - // If no full-state op, we need to reset state before applying incremental ops. - // This is because USE_REMOTE should REPLACE local state with remote state, - // not merge remote ops into existing local state. - if (!hasFullStateOp) { - OpLog.normal( - 'OperationLogSyncService: No full-state op in remote. Resetting state before applying incremental ops.', - ); - // Reset to default/empty state so incremental ops build fresh state - const defaultData = getDefaultMainModelData(); - this.store.dispatch( - loadAllData({ - appDataComplete: defaultData as Parameters< - typeof loadAllData - >[0]['appDataComplete'], - }), - ); - // Brief yield to let NgRx process the state reset - await new Promise((resolve) => setTimeout(resolve, 0)); - } - - // Process all remote ops (no confirmation needed - user already chose USE_REMOTE). - // Skip conflict detection because the NgRx store was just reset to empty state, - // which causes all entities to appear missing and CONCURRENT ops to be discarded. - // Validation failure is surfaced via the session-validation latch. (#7330) - await this.remoteOpsProcessingService.processRemoteOps(result.newOps, { - skipConflictDetection: true, - }); - - // Update lastServerSeq - if (result.latestServerSeq !== undefined) { - await syncProvider.setLastServerSeq(result.latestServerSeq); - } + // Update lastServerSeq + if (result.latestServerSeq !== undefined) { + await syncProvider.setLastServerSeq(result.latestServerSeq); } OpLog.normal( - `OperationLogSyncService: Force download complete. Processed ${result.newOps.length} ops.`, + `OperationLogSyncService: Force download complete. Rebuilt from ${result.newOps.length} ops.`, ); this._showRestorePreviousDataSnack(backupSavedAt); } diff --git a/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts b/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts index 453032f51e..11a1badebb 100644 --- a/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts +++ b/src/app/op-log/sync/operation-log-upload-piggyback-seq.integration.spec.ts @@ -191,6 +191,7 @@ describe('OperationLogSyncService + OperationLogUploadService — piggyback seq allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }); dialogServiceSpy = jasmine.createSpyObj('SyncImportConflictDialogService', [ @@ -393,6 +394,7 @@ describe('OperationLogSyncService + OperationLogUploadService — piggyback seq allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: false, }; }); setLastServerSeqSpy.and.callFake(async (n: number) => { diff --git a/src/app/op-log/sync/operation-log-upload.service.spec.ts b/src/app/op-log/sync/operation-log-upload.service.spec.ts index 46dbe09b30..8a9a5fac18 100644 --- a/src/app/op-log/sync/operation-log-upload.service.spec.ts +++ b/src/app/op-log/sync/operation-log-upload.service.spec.ts @@ -1016,7 +1016,7 @@ describe('OperationLogUploadService', () => { }); it('should mark regular ops as synced when full-state op is uploaded (ops before snapshot)', async () => { - // Regular op id 'op-0' sorts BEFORE full-state op id 'op-1', + // Regular op seq 1 is BEFORE full-state op seq 2, // meaning the regular op was created before the snapshot and is included in it. const regularEntry = createMockEntry(1, 'op-0', 'client-1'); const fullStateEntry = createFullStateEntry( @@ -1042,7 +1042,7 @@ describe('OperationLogUploadService', () => { }); it('should mark regular ops as synced when Repair op is uploaded (ops before snapshot)', async () => { - // Regular op id 'op-0' sorts BEFORE full-state op id 'op-1' + // Regular op seq 1 is BEFORE full-state op seq 2 const regularEntry = createMockEntry(1, 'op-0', 'client-1'); const fullStateEntry = createFullStateEntry(2, 'op-1', 'client-1', OpType.Repair); mockOpLogStore.getUnsynced.and.returnValue( @@ -1062,7 +1062,7 @@ describe('OperationLogUploadService', () => { }); it('should upload regular ops created AFTER full-state snapshot', async () => { - // Full-state op id 'op-1' sorts BEFORE regular op id 'op-2', + // Full-state op seq 1 is BEFORE regular op seq 2, // meaning the regular op was created AFTER the snapshot and is NOT included in it. const fullStateEntry = createFullStateEntry( 1, @@ -1094,6 +1094,40 @@ describe('OperationLogUploadService', () => { expect(mockOpLogStore.markSynced).toHaveBeenCalledWith([2]); }); + it('should upload a post-snapshot op even when its UUIDv7 id sorts before the full-state op id (clock rollback)', async () => { + // Wall-clock rollback regression: the regular op was created AFTER the + // snapshot (seq 2 > seq 1) but got a lexically SMALLER UUIDv7 id + // ('op-0' < 'op-1'). It is NOT in the frozen snapshot payload, so it + // must be uploaded — never just marked synced. + const fullStateEntry = createFullStateEntry( + 1, + 'op-1', + 'client-1', + OpType.BackupImport, + ); + const regularEntry = createMockEntry(2, 'op-0', 'client-1'); + mockOpLogStore.getUnsynced.and.returnValue( + Promise.resolve([fullStateEntry, regularEntry]), + ); + mockApiProvider.uploadOps.and.returnValue( + Promise.resolve({ + results: [{ opId: 'op-0', accepted: true }], + latestSeq: 2, + newOps: [], + }), + ); + + const result = await service.uploadPendingOps(mockApiProvider); + + expect(mockApiProvider.uploadSnapshot).toHaveBeenCalled(); + expect(mockApiProvider.uploadOps).toHaveBeenCalled(); + const uploadedOpIds = mockApiProvider.uploadOps.calls + .mostRecent() + .args[0].map((op) => op.id); + expect(uploadedOpIds).toEqual(['op-0']); + expect(result.uploadedCount).toBe(2); + }); + it('should NOT auto-set isCleanSlate for SyncImport unlike BackupImport/Repair', async () => { const entry = createFullStateEntry(1, 'op-1', 'client-1', OpType.SyncImport); mockOpLogStore.getUnsynced.and.returnValue(Promise.resolve([entry])); diff --git a/src/app/op-log/sync/operation-log-upload.service.ts b/src/app/op-log/sync/operation-log-upload.service.ts index a84d79d977..93bef79ca4 100644 --- a/src/app/op-log/sync/operation-log-upload.service.ts +++ b/src/app/op-log/sync/operation-log-upload.service.ts @@ -182,7 +182,10 @@ export class OperationLogUploadService { // Upload full-state operations via snapshot endpoint let fullStateOpUploaded = false; - let lastUploadedFullStateOpId: string | undefined; + // Local append order (seq), not op id: UUIDv7 ids follow the wall clock and + // can order a post-snapshot op BEFORE the full-state op after a clock + // rollback, which would mark it synced without ever uploading it. + let lastUploadedFullStateOpSeq: number | undefined; for (const entry of fullStateOps) { // BackupImport/Repair: always wipe server (recovery operations replace all state) // SyncImport: only wipe when explicitly requested (preserves SYNC_IMPORT_EXISTS check) @@ -202,9 +205,10 @@ export class OperationLogUploadService { lastKnownServerSeq = result.serverSeq; highestReceivedSeq = Math.max(highestReceivedSeq, result.serverSeq); } - // Track that a full-state op was uploaded - regular ops before it are already included + // Track that a full-state op was uploaded - regular ops appended before it + // are already included in its frozen snapshot payload fullStateOpUploaded = true; - lastUploadedFullStateOpId = entry.op.id; + lastUploadedFullStateOpSeq = entry.seq; } else { // Special handling for SYNC_IMPORT_EXISTS: another client already uploaded // a SYNC_IMPORT. We should delete our local SYNC_IMPORT and let the normal @@ -245,11 +249,11 @@ export class OperationLogUploadService { return; } - if (fullStateOpUploaded && lastUploadedFullStateOpId) { + if (fullStateOpUploaded && lastUploadedFullStateOpSeq !== undefined) { const { opsIncludedInSnapshot, opsAfterSnapshot } = planRegularOpsAfterFullStateUpload({ regularOps, - lastUploadedFullStateOpId, + lastUploadedFullStateOpSeq, }); if (opsIncludedInSnapshot.length > 0) { diff --git a/src/app/op-log/sync/remote-ops-processing.service.spec.ts b/src/app/op-log/sync/remote-ops-processing.service.spec.ts index 0392b7ae3d..c59653fc0f 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.spec.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.spec.ts @@ -4,7 +4,6 @@ import { of } from 'rxjs'; import { RemoteOpsProcessingService } from './remote-ops-processing.service'; import { SchemaMigrationService, - MAX_VERSION_SKIP, MIN_SUPPORTED_SCHEMA_VERSION, } from '../persistence/schema-migration.service'; import { SnackService } from '../../core/snack/snack.service'; @@ -400,7 +399,7 @@ describe('RemoteOpsProcessingService', () => { ); }); - it('should skip ops that throw during migration but continue processing others', async () => { + it('should stop the batch at the first op that throws during migration and only process the prefix', async () => { const remoteOps: Operation[] = [ { id: 'op1', schemaVersion: 1 } as Operation, { id: 'throws', schemaVersion: 1 } as Operation, @@ -419,17 +418,26 @@ describe('RemoteOpsProcessingService', () => { vectorClockServiceSpy.getSnapshotVectorClock.and.returnValue(Promise.resolve({})); opLogStoreSpy.hasOp.and.returnValue(Promise.resolve(false)); - await service.processRemoteOps(remoteOps); + const result = await service.processRemoteOps(remoteOps); - // op1 and op3 should be processed (appendBatchSkipDuplicates called with array of both) + // Only the prefix before the blocked op is processed. op3 must NOT be + // applied: it may depend on the blocked op, and the un-advanced cursor + // will re-deliver it after the migration issue is fixed. expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith( - [remoteOps[0], remoteOps[2]], + [remoteOps[0]], 'remote', { pendingApply: true }, ); + expect(result.blockedByIncompatibleOp).toBe(true); + expect(snackServiceSpy.open).toHaveBeenCalledWith( + jasmine.objectContaining({ + type: 'ERROR', + msg: T.F.SYNC.S.MIGRATION_FAILED, + }), + ); }); - it('should return early when all ops fail migration', async () => { + it('should block without processing anything when the first op fails migration', async () => { const remoteOps: Operation[] = [ { id: 'op1', schemaVersion: 1 } as Operation, { id: 'op2', schemaVersion: 1 } as Operation, @@ -446,6 +454,7 @@ describe('RemoteOpsProcessingService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, }); }); @@ -487,12 +496,13 @@ describe('RemoteOpsProcessingService', () => { ); }); - it('should show error snackbar and abort if version is too new', async () => { - const remoteOps: Operation[] = [ - { id: 'op1', schemaVersion: 1 + MAX_VERSION_SKIP + 1 } as Operation, - ]; + it('should block any op from a newer schema version (no forward-compat band)', async () => { + // Current version is 1 (set in beforeEach). Even version 2 — one ahead — + // must block: real migrations rename/split fields, so a future op applied + // verbatim corrupts state. + const remoteOps: Operation[] = [{ id: 'op1', schemaVersion: 2 } as Operation]; - await service.processRemoteOps(remoteOps); + const result = await service.processRemoteOps(remoteOps); expect(snackServiceSpy.open).toHaveBeenCalledWith( jasmine.objectContaining({ @@ -503,6 +513,30 @@ describe('RemoteOpsProcessingService', () => { // Should not proceed to apply ops expect(opLogStoreSpy.getUnsynced).not.toHaveBeenCalled(); + expect(result.blockedByIncompatibleOp).toBe(true); + }); + + it('should process the prefix before a too-new op but flag the block', async () => { + const remoteOps: Operation[] = [ + { id: 'op1', schemaVersion: 1 } as Operation, + { id: 'future', schemaVersion: 2 } as Operation, + { id: 'op3', schemaVersion: 1 } as Operation, + ]; + + opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([])); + opLogStoreSpy.getUnsyncedByEntity.and.returnValue(Promise.resolve(new Map())); + vectorClockServiceSpy.getEntityFrontier.and.returnValue(Promise.resolve(new Map())); + vectorClockServiceSpy.getSnapshotVectorClock.and.returnValue(Promise.resolve({})); + opLogStoreSpy.hasOp.and.returnValue(Promise.resolve(false)); + + const result = await service.processRemoteOps(remoteOps); + + expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith( + [remoteOps[0]], + 'remote', + { pendingApply: true }, + ); + expect(result.blockedByIncompatibleOp).toBe(true); }); it('should show error snackbar and abort if version is below minimum supported', async () => { @@ -526,69 +560,26 @@ describe('RemoteOpsProcessingService', () => { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: true, }); }); - it('should show warning once per session when receiving ops from newer version', async () => { - // Current version is 1 (set in beforeEach) - const remoteOps: Operation[] = [ - { id: 'op1', schemaVersion: 2 } as Operation, - { id: 'op2', schemaVersion: 2 } as Operation, - ]; - - // Setup for processing - opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([])); - opLogStoreSpy.getUnsyncedByEntity.and.returnValue(Promise.resolve(new Map())); - vectorClockServiceSpy.getEntityFrontier.and.returnValue(Promise.resolve(new Map())); - vectorClockServiceSpy.getSnapshotVectorClock.and.returnValue(Promise.resolve({})); - opLogStoreSpy.hasOp.and.returnValue(Promise.resolve(false)); - opLogStoreSpy.append.and.returnValue(Promise.resolve(1)); - - await service.processRemoteOps(remoteOps); - - // Should show warning exactly once (not twice for two ops) - expect(snackServiceSpy.open).toHaveBeenCalledTimes(1); - expect(snackServiceSpy.open).toHaveBeenCalledWith( - jasmine.objectContaining({ - type: 'WARNING', - msg: T.F.SYNC.S.NEWER_VERSION_AVAILABLE, - }), - ); - - // Should still process the ops (appendBatchSkipDuplicates called with both ops) - expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith( - remoteOps, - 'remote', - { - pendingApply: true, - }, - ); - }); - - it('should not show newer version warning again in same session', async () => { + it('should show the version-block error only once per session', async () => { // Current version is 1 (set in beforeEach) const remoteOps1: Operation[] = [{ id: 'op1', schemaVersion: 2 } as Operation]; const remoteOps2: Operation[] = [{ id: 'op2', schemaVersion: 2 } as Operation]; - // Setup for processing - opLogStoreSpy.getUnsynced.and.returnValue(Promise.resolve([])); - opLogStoreSpy.getUnsyncedByEntity.and.returnValue(Promise.resolve(new Map())); - vectorClockServiceSpy.getEntityFrontier.and.returnValue(Promise.resolve(new Map())); - vectorClockServiceSpy.getSnapshotVectorClock.and.returnValue(Promise.resolve({})); - opLogStoreSpy.hasOp.and.returnValue(Promise.resolve(false)); - opLogStoreSpy.append.and.returnValue(Promise.resolve(1)); - // First call await service.processRemoteOps(remoteOps1); - // Second call (same session) + // Second call (same session — periodic sync retries re-hit the block) await service.processRemoteOps(remoteOps2); - // Warning should only be shown once across both calls + // Error snack should only be shown once across both calls expect(snackServiceSpy.open).toHaveBeenCalledTimes(1); expect(snackServiceSpy.open).toHaveBeenCalledWith( jasmine.objectContaining({ - type: 'WARNING', - msg: T.F.SYNC.S.NEWER_VERSION_AVAILABLE, + type: 'ERROR', + msg: T.F.SYNC.S.VERSION_TOO_OLD, }), ); }); diff --git a/src/app/op-log/sync/remote-ops-processing.service.ts b/src/app/op-log/sync/remote-ops-processing.service.ts index 576e0e5b61..d1e910f3c9 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.ts @@ -19,7 +19,6 @@ import { ValidateStateService } from '../validation/validate-state.service'; import { SyncSessionValidationService } from './sync-session-validation.service'; import { VectorClockService } from './vector-clock.service'; import { - MAX_VERSION_SKIP, MIN_SUPPORTED_SCHEMA_VERSION, SchemaMigrationService, } from '../persistence/schema-migration.service'; @@ -69,8 +68,8 @@ export class RemoteOpsProcessingService { private writeFlushService = inject(OperationWriteFlushService); private injector = inject(Injector); - /** Flag to show newer version warning only once per session */ - private _hasWarnedNewerVersionThisSession = false; + /** Flag to show version-incompatibility warnings only once per session */ + private _hasWarnedVersionBlockThisSession = false; /** Flag to show migration failure warning only once per session */ private _hasWarnedMigrationFailureThisSession = false; @@ -100,6 +99,7 @@ export class RemoteOpsProcessingService { filteredOpCount: number; filteringImport?: Operation; isLocalUnsyncedImport: boolean; + blockedByIncompatibleOp: boolean; }> { // Validation failure surfaces via the SyncSessionValidationService latch // (#7330). `validateAfterSync` and the conflict-resolution validation path @@ -109,52 +109,41 @@ export class RemoteOpsProcessingService { // ───────────────────────────────────────────────────────────────────────── // STEP 1: Schema Migration (Receiver-Side) // Migrate ops from older schema versions to current version. - // - Ops below MIN_SUPPORTED_SCHEMA_VERSION: error, stop sync - // - Ops beyond MAX_VERSION_SKIP: error, stop sync - // - Ops from newer version (within skip): warning once per session, continue + // + // Blocking semantics: an op that cannot be terminally processed here — + // below MIN_SUPPORTED_SCHEMA_VERSION, from a NEWER schema version (real + // migrations rename/split fields, so a future op applied verbatim corrupts + // state), or throwing during migration — STOPS the batch at that op. Ops + // before the block are processed normally; the blocked op and everything + // after are neither stored nor applied. Callers must NOT advance the + // server cursor when `blockedByIncompatibleOp` is true, so the blocked op + // is re-downloaded and retried after an app update / migration fix instead + // of being skipped forever (already-processed prefix ops are deduplicated + // by the appliedOpIds download filter). Ops migrated to `null` are an + // intentional terminal drop and do NOT block. // ───────────────────────────────────────────────────────────────────────── const currentVersion = this.schemaMigrationService.getCurrentVersion(); const migratedOps: Operation[] = []; const droppedEntityIds = new Set(); - const failedMigrationOpIds: string[] = []; - let updateRequired = false; + let blockReason: 'VERSION_UNSUPPORTED' | 'VERSION_TOO_NEW' | 'MIGRATION_FAILED' = + 'MIGRATION_FAILED'; + let blockedOp: Operation | null = null; for (const op of remoteOps) { const opVersion = op.schemaVersion ?? 1; - // Check if remote op is too old (below minimum supported) + // Op below minimum supported version: no migration path exists. if (opVersion < MIN_SUPPORTED_SCHEMA_VERSION) { - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.VERSION_UNSUPPORTED, - actionStr: T.PS.UPDATE_APP, - actionFn: () => - window.open('https://super-productivity.com/download', '_blank'), - }); - return { - localWinOpsCreated: 0, - allOpsFilteredBySyncImport: false, - filteredOpCount: 0, - isLocalUnsyncedImport: false, - }; - } - - // Check if remote op is too new (exceeds supported skip) - if (opVersion > currentVersion + MAX_VERSION_SKIP) { - updateRequired = true; + blockedOp = op; + blockReason = 'VERSION_UNSUPPORTED'; break; } - // Warn once per session if receiving ops from a newer version - if (opVersion > currentVersion && !this._hasWarnedNewerVersionThisSession) { - this._hasWarnedNewerVersionThisSession = true; - this.snackService.open({ - type: 'WARNING', - msg: T.F.SYNC.S.NEWER_VERSION_AVAILABLE, - actionStr: T.PS.UPDATE_APP, - actionFn: () => - window.open('https://super-productivity.com/download', '_blank'), - }); + // Op from a newer schema version: this client cannot interpret it safely. + if (opVersion > currentVersion) { + blockedOp = op; + blockReason = 'VERSION_TOO_NEW'; + break; } try { @@ -178,44 +167,23 @@ export class RemoteOpsProcessingService { } } catch (e) { OpLog.err(`RemoteOpsProcessingService: Migration failed for op ${op.id}`, e); - // Track failed migrations to notify user. If ops are from a compatible version, - // this indicates a bug or data corruption. - failedMigrationOpIds.push(op.id); + blockedOp = op; + blockReason = 'MIGRATION_FAILED'; + break; } } - // Notify user if any migrations failed (once per session to avoid spam) - if (failedMigrationOpIds.length > 0) { - OpLog.warn( - `RemoteOpsProcessingService: ${failedMigrationOpIds.length} op(s) failed migration`, - { failedOpIds: failedMigrationOpIds }, + if (blockedOp) { + OpLog.err( + `RemoteOpsProcessingService: Blocked at op ${blockedOp.id} (${blockReason}, ` + + `schemaVersion=${blockedOp.schemaVersion ?? 1}, current=${currentVersion}). ` + + 'Processing the batch prefix only; cursor must not advance past this op.', ); - if (!this._hasWarnedMigrationFailureThisSession) { - this._hasWarnedMigrationFailureThisSession = true; - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.MIGRATION_FAILED, - }); - } - } - - if (updateRequired) { - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.VERSION_TOO_OLD, - actionStr: T.PS.UPDATE_APP, - actionFn: () => window.open('https://super-productivity.com/download', '_blank'), - }); - return { - localWinOpsCreated: 0, - allOpsFilteredBySyncImport: false, - filteredOpCount: 0, - isLocalUnsyncedImport: false, - }; + this._notifyBlockedOp(blockReason); } if (migratedOps.length === 0) { - if (remoteOps.length > 0) { + if (remoteOps.length > 0 && !blockedOp) { OpLog.normal( 'RemoteOpsProcessingService: All remote ops were dropped during migration.', ); @@ -225,8 +193,10 @@ export class RemoteOpsProcessingService { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp: blockedOp !== null, }; } + const blockedByIncompatibleOp = blockedOp !== null; // ───────────────────────────────────────────────────────────────────────── // STEP 2: Filter ops invalidated by SYNC_IMPORT @@ -258,6 +228,7 @@ export class RemoteOpsProcessingService { filteredOpCount: invalidatedOps.length, filteringImport, isLocalUnsyncedImport, + blockedByIncompatibleOp, }; } @@ -285,6 +256,7 @@ export class RemoteOpsProcessingService { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp, }; } @@ -316,6 +288,7 @@ export class RemoteOpsProcessingService { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp, }; } @@ -376,9 +349,42 @@ export class RemoteOpsProcessingService { allOpsFilteredBySyncImport: false, filteredOpCount: 0, isLocalUnsyncedImport: false, + blockedByIncompatibleOp, }; } + /** + * User notification for a version/migration block, once per session per + * category to avoid snack spam from periodic sync retries (the block persists + * until an app update or migration fix, and every retry re-hits it). + */ + private _notifyBlockedOp( + reason: 'VERSION_UNSUPPORTED' | 'VERSION_TOO_NEW' | 'MIGRATION_FAILED', + ): void { + if (reason === 'MIGRATION_FAILED') { + if (!this._hasWarnedMigrationFailureThisSession) { + this._hasWarnedMigrationFailureThisSession = true; + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.MIGRATION_FAILED, + }); + } + return; + } + if (!this._hasWarnedVersionBlockThisSession) { + this._hasWarnedVersionBlockThisSession = true; + this.snackService.open({ + type: 'ERROR', + msg: + reason === 'VERSION_UNSUPPORTED' + ? T.F.SYNC.S.VERSION_UNSUPPORTED + : T.F.SYNC.S.VERSION_TOO_OLD, + actionStr: T.PS.UPDATE_APP, + actionFn: () => window.open('https://super-productivity.com/download', '_blank'), + }); + } + } + /** * Applies non-conflicting operations with crash-safe tracking. * diff --git a/src/app/op-log/sync/sync-import-conflict-gate.service.spec.ts b/src/app/op-log/sync/sync-import-conflict-gate.service.spec.ts index d833226618..2718eeddb4 100644 --- a/src/app/op-log/sync/sync-import-conflict-gate.service.spec.ts +++ b/src/app/op-log/sync/sync-import-conflict-gate.service.spec.ts @@ -333,4 +333,131 @@ describe('SyncImportConflictGateService', () => { expect(writeFlushServiceSpy.flushPendingWrites).not.toHaveBeenCalled(); expect(opLogStoreSpy.getUnsynced).toHaveBeenCalled(); }); + + describe('meaningful-work coverage beyond TASK/PROJECT/TAG/NOTE CUD', () => { + it('should treat a pending MOV op as meaningful', async () => { + const pendingMove = createEntry( + createOperation({ + id: 'local-move', + actionType: '[Task] Move task' as ActionType, + opType: OpType.Move, + entityType: 'TASK', + entityId: 'task-1', + clientId: 'client-A', + vectorClock: { clientA: 1 }, + }), + ); + opLogStoreSpy.getUnsynced.and.resolveTo([pendingMove]); + + const result = await service.checkIncomingFullStateConflict([createOperation()]); + + expect(result.hasMeaningfulPending).toBeTrue(); + expect(result.dialogData).toBeDefined(); + }); + + it('should treat pending non-task entity work (TIME_TRACKING, SIMPLE_COUNTER) as meaningful', async () => { + const pendingTimeTracking = createEntry( + createOperation({ + id: 'local-tt', + actionType: '[TimeTracking] Update whole day' as ActionType, + opType: OpType.Update, + entityType: 'TIME_TRACKING', + entityId: 'tt-1', + clientId: 'client-A', + vectorClock: { clientA: 1 }, + }), + ); + const pendingCounter = createEntry( + createOperation({ + id: 'local-counter', + actionType: '[SimpleCounter] Increase Counter Today' as ActionType, + opType: OpType.Update, + entityType: 'SIMPLE_COUNTER', + entityId: 'counter-1', + clientId: 'client-A', + vectorClock: { clientA: 2 }, + }), + ); + + expect(service.hasMeaningfulPendingOps([pendingTimeTracking])).toBeTrue(); + expect(service.hasMeaningfulPendingOps([pendingCounter])).toBeTrue(); + }); + + it('should still treat MIGRATION/RECOVERY bookkeeping ops as not meaningful', async () => { + const pendingMigration = createEntry( + createOperation({ + id: 'local-genesis', + actionType: '[Migration] Genesis' as ActionType, + opType: OpType.Create, + entityType: 'MIGRATION', + entityId: 'genesis', + clientId: 'client-A', + vectorClock: { clientA: 1 }, + }), + ); + + expect(service.hasMeaningfulPendingOps([pendingMigration])).toBeFalse(); + }); + }); + + describe('preCapturedPendingOps (piggyback-upload race)', () => { + it('should judge meaningfulness against the pre-upload snapshot, not the live pending set', async () => { + // Live pending set is empty — the op was accepted and marked synced during + // the same upload round that piggybacked the import. + opLogStoreSpy.getUnsynced.and.resolveTo([]); + + const acceptedThisRound = createEntry( + createOperation({ + id: 'accepted-op', + actionType: '[Task] Update task' as ActionType, + opType: OpType.Update, + entityType: 'TASK', + entityId: 'task-1', + clientId: 'client-A', + vectorClock: { clientA: 1 }, + }), + ); + + const result = await service.checkIncomingFullStateConflict([createOperation()], { + preCapturedPendingOps: [acceptedThisRound], + }); + + expect(result.hasMeaningfulPending).toBeTrue(); + expect(result.dialogData).toBeDefined(); + }); + + it('should derive discardable example-task ids from the LIVE pending set', async () => { + const exampleCreate = (id: string): OperationLogEntry => + createEntry( + createOperation({ + id, + actionType: ActionType.TASK_SHARED_ADD, + opType: OpType.Create, + entityType: 'TASK', + entityId: `task-${id}`, + payload: { + actionPayload: { + task: { id: `task-${id}` }, + isExampleTask: true, + }, + entityChanges: [], + }, + clientId: 'client-A', + vectorClock: { clientA: 1 }, + }), + ); + + // Pre-captured snapshot has two example ops; one was accepted (synced) + // during the round, so only the still-pending one may be discarded. + const stillPending = exampleCreate('example-still-pending'); + const acceptedAlready = exampleCreate('example-accepted'); + opLogStoreSpy.getUnsynced.and.resolveTo([stillPending]); + + const result = await service.checkIncomingFullStateConflict([createOperation()], { + preCapturedPendingOps: [stillPending, acceptedAlready], + }); + + expect(result.discardablePendingOpIds).toEqual(['example-still-pending']); + }); + }); }); diff --git a/src/app/op-log/sync/sync-import-conflict-gate.service.ts b/src/app/op-log/sync/sync-import-conflict-gate.service.ts index b8013aee3a..bc255685de 100644 --- a/src/app/op-log/sync/sync-import-conflict-gate.service.ts +++ b/src/app/op-log/sync/sync-import-conflict-gate.service.ts @@ -10,7 +10,16 @@ import { OperationWriteFlushService } from './operation-write-flush.service'; import { SyncImportConflictData } from './dialog-sync-import-conflict/dialog-sync-import-conflict.component'; import { isExampleTaskCreateOp } from '../validation/is-example-task-op.util'; -const USER_ENTITY_TYPES = new Set(['TASK', 'PROJECT', 'TAG', 'NOTE']); +/** + * Pending ops on these entity types are startup/system noise, not user work: + * config writes happen automatically (defaults, migrations) and sync settings + * are intentionally local; MIGRATION/RECOVERY are bookkeeping genesis ops. + * Everything else — including MOV/BATCH ops and entities like TIME_TRACKING, + * SIMPLE_COUNTER, TASK_REPEAT_CFG, PLANNER, BOARD — is user work an incoming + * full-state import would silently discard (imports drop concurrent ops by + * design), so it must count as meaningful and trigger the conflict dialog. + */ +const DISCARDABLE_ENTITY_TYPES = new Set(['GLOBAL_CONFIG', 'MIGRATION', 'RECOVERY']); export interface IncomingFullStateConflictGateResult { fullStateOp?: Operation; @@ -36,9 +45,10 @@ export class SyncImportConflictGateService { private writeFlushService = inject(OperationWriteFlushService); /** - * Config-only pending ops are not considered user work for this conflict gate. - * Full-state ops are always meaningful because applying a newer full-state op can - * invalidate their local import/repair semantics. + * Every pending op is user work unless it is on a DISCARDABLE_ENTITY_TYPES + * entity or is an onboarding example-task create. Full-state ops are always + * meaningful because applying a newer full-state op can invalidate their + * local import/repair semantics. */ hasMeaningfulPendingOps(ops: OperationLogEntry[]): boolean { return ops.some((entry) => { @@ -48,18 +58,24 @@ export class SyncImportConflictGateService { if (isExampleTaskCreateOp(entry)) { return false; } - return ( - USER_ENTITY_TYPES.has(entry.op.entityType) && - (entry.op.opType === OpType.Create || - entry.op.opType === OpType.Update || - entry.op.opType === OpType.Delete) - ); + return !DISCARDABLE_ENTITY_TYPES.has(entry.op.entityType); }); } + /** + * @param options.preCapturedPendingOps - Pending ops captured BEFORE the upload + * round started. The piggyback-upload path MUST pass this: by the time + * its gate runs, ops accepted in the same round were already marked + * synced, so a live getUnsynced() read would no longer see local work + * that the piggybacked import is about to discard. + */ async checkIncomingFullStateConflict( incomingOps: Operation[], - options: { flushPendingWrites?: boolean; isNeverSynced?: boolean } = {}, + options: { + flushPendingWrites?: boolean; + isNeverSynced?: boolean; + preCapturedPendingOps?: OperationLogEntry[]; + } = {}, ): Promise { const fullStateOp = incomingOps.find((op) => FULL_STATE_OP_TYPES.has(op.opType)); @@ -78,13 +94,21 @@ export class SyncImportConflictGateService { await this.writeFlushService.flushPendingWrites(); } - const pendingOps = await this.opLogStore.getUnsynced(); + const pendingOps = + options.preCapturedPendingOps ?? (await this.opLogStore.getUnsynced()); const hasMeaningfulPending = this.hasMeaningfulPendingOps(pendingOps); // Example-task ops that the caller may reject when it accepts the import silently. // When `hasMeaningfulPending` is true (real work pending alongside example tasks), // the conflict dialog is shown instead and these are intentionally left untouched: // if the user keeps local state, their example tasks ride along with the rest. - const discardablePendingOpIds = pendingOps + // + // These must come from a LIVE read: with a pre-captured snapshot, example ops + // accepted earlier in the same upload round are already marked synced and must + // not be re-marked rejected by the caller. + const livePendingOps = options.preCapturedPendingOps + ? await this.opLogStore.getUnsynced() + : pendingOps; + const discardablePendingOpIds = livePendingOps .filter(isExampleTaskCreateOp) .map((entry) => entry.op.id); diff --git a/src/app/op-log/testing/integration/migration-handling.integration.spec.ts b/src/app/op-log/testing/integration/migration-handling.integration.spec.ts index aaf7460e00..f6d303695c 100644 --- a/src/app/op-log/testing/integration/migration-handling.integration.spec.ts +++ b/src/app/op-log/testing/integration/migration-handling.integration.spec.ts @@ -1,10 +1,7 @@ import { TestBed } from '@angular/core/testing'; import { RemoteOpsProcessingService } from '../../sync/remote-ops-processing.service'; import { OperationLogStoreService } from '../../persistence/operation-log-store.service'; -import { - SchemaMigrationService, - MAX_VERSION_SKIP, -} from '../../persistence/schema-migration.service'; +import { SchemaMigrationService } from '../../persistence/schema-migration.service'; import { SnackService } from '../../../core/snack/snack.service'; import { VectorClockService } from '../../sync/vector-clock.service'; import { OperationApplierService } from '../../apply/operation-applier.service'; @@ -146,28 +143,14 @@ describe('Migration Handling Integration', () => { ); }); - it('should accept operation with compatible future version (within skip limit)', async () => { - // Logic: if opVersion <= current + MAX_VERSION_SKIP, it's accepted - // MAX_VERSION_SKIP is 3. So current + 3 should still be accepted. - // Note: A WARNING snackbar may be shown for newer versions, but no ERROR. - const compatibleVersion = CURRENT_SCHEMA_VERSION + MAX_VERSION_SKIP; - const op = createOp(compatibleVersion); + it('should block operation from any future version (no forward-compat band)', async () => { + // Forward-compatible migrations are not implemented: real migrations + // rename/split fields, so a future op applied verbatim corrupts state. + // Even one version ahead must block until the app is updated. + const futureVersion = CURRENT_SCHEMA_VERSION + 1; + const op = createOp(futureVersion); - await service.processRemoteOps([op]); - - // Should NOT show error (operation is accepted) - expect(snackServiceSpy.open).not.toHaveBeenCalledWith( - jasmine.objectContaining({ type: 'ERROR' }), - ); - expect(operationApplierSpy.applyOperations).toHaveBeenCalled(); - }); - - it('should reject operation with incompatible future version', async () => { - // Logic: if opVersion > current + MAX_VERSION_SKIP, update required - const incompatibleVersion = CURRENT_SCHEMA_VERSION + MAX_VERSION_SKIP + 1; - const op = createOp(incompatibleVersion); - - await service.processRemoteOps([op]); + const result = await service.processRemoteOps([op]); // Should trigger error snackbar expect(snackServiceSpy.open).toHaveBeenCalledWith( @@ -177,8 +160,9 @@ describe('Migration Handling Integration', () => { }), ); - // Should NOT apply operation + // Should NOT apply operation, and callers must hold the cursor expect(operationApplierSpy.applyOperations).not.toHaveBeenCalled(); + expect(result.blockedByIncompatibleOp).toBe(true); }); it('should handle operations missing schemaVersion (default to 1)', async () => {