mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-18 00:46:45 +00:00
fix(sync): remediate sync-correctness review findings
Address findings from a full sync-system review (blockers + high/medium):
- USE_REMOTE ("Use Server Data") is now a true download-first rebuild:
fetches the complete server history (incl. own/already-known ops via a
raw-download mode) and validates it before any local mutation; aborts
untouched on download failure, empty remote, or newer-schema ops.
- Widen the incoming full-state conflict gate to treat all user work as
meaningful (not just TASK/PROJECT/TAG/NOTE CUD); fix the piggyback race
by judging against a pre-upload pending snapshot.
- Stop advancing the server cursor past ops blocked by schema/migration
failures so they are retried after an app update; block any op from a
newer schema version (no forward-compat band).
- Retry of failed remote ops runs archive side effects only, never
re-dispatching reducers whose effects already committed (fixes additive
double-apply of time-tracking/counter deltas across hydration+retry).
- Use local monotonic seq (not lexical UUIDv7) to decide which ops are
covered by an uploaded snapshot, preventing dropped unsynced ops under
clock rollback.
- Server: include entityIds in duplicate-operation equality.
- Capture buffer no longer silently drops accepted actions at 100 (warns
to reload); drop only past a 5000 pathological cap.
Adds regression coverage for each fix. sync-core 209/209,
super-sync-server 817/817, and all touched Angular specs pass.
This commit is contained in:
Johannes Millan
parent
6e20c09e25
commit
79f91e36fe
31 changed files with 1232 additions and 316 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -45,7 +45,7 @@ android/.idea
|
|||
# misc
|
||||
\\backups
|
||||
/backups
|
||||
/.angular/cache
|
||||
/.angular
|
||||
/.sass-cache
|
||||
/connect.lock
|
||||
/coverage
|
||||
|
|
|
|||
|
|
@ -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 &&
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -93,6 +93,7 @@ const createStoredDuplicateOp = (op: ReturnType<typeof createOp>) => ({
|
|||
opType: op.opType,
|
||||
entityType: op.entityType,
|
||||
entityId: op.entityId,
|
||||
entityIds: [],
|
||||
payload: op.payload,
|
||||
vectorClock: op.vectorClock,
|
||||
schemaVersion: op.schemaVersion,
|
||||
|
|
|
|||
|
|
@ -12,7 +12,11 @@ export interface ApplyOperationsResult<TOperation extends Operation<string> = 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string> = Operation,
|
||||
|
|
|
|||
|
|
@ -71,7 +71,9 @@ export const yieldToEventLoop = (): Promise<void> =>
|
|||
* 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({
|
||||
|
|
|
|||
|
|
@ -11,24 +11,33 @@ export interface PlanRegularOpsAfterFullStateUploadOptions<
|
|||
TEntry extends OperationLogEntry<Operation<string>> = 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<Operation<string>> = OperationLogEntry,
|
||||
>({
|
||||
regularOps,
|
||||
lastUploadedFullStateOpId,
|
||||
lastUploadedFullStateOpSeq,
|
||||
}: PlanRegularOpsAfterFullStateUploadOptions<TEntry>): RegularOpsAfterFullStateUploadPlan<TEntry> => {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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<BulkReplayAction> = {
|
||||
dispatch: vi.fn(() => {
|
||||
callOrder.push('dispatchBulk');
|
||||
}),
|
||||
};
|
||||
const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
|
||||
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<BulkReplayAction> = { dispatch: vi.fn() };
|
||||
const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
|
||||
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<ReplayAction> = {
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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)', () => {
|
||||
|
|
|
|||
|
|
@ -106,7 +106,10 @@ export class OperationApplierService implements OperationApplyPort<Operation> {
|
|||
|
||||
const result = await replayOperationBatch({
|
||||
ops,
|
||||
applyOptions: { isLocalHydration },
|
||||
applyOptions: {
|
||||
isLocalHydration,
|
||||
skipReducerDispatch: options.skipReducerDispatch,
|
||||
},
|
||||
dispatcher: this.store,
|
||||
createBulkApplyAction: (operations) =>
|
||||
bulkApplyOperations({ operations, localClientId }),
|
||||
|
|
|
|||
|
|
@ -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)', () => {
|
||||
|
|
|
|||
|
|
@ -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`,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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`,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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<DownloadResult> {
|
||||
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<DownloadResult> {
|
||||
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<string>()
|
||||
: 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}`,
|
||||
|
|
|
|||
|
|
@ -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<OperationLogDownloadService>;
|
||||
|
||||
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<OperationLogDownloadService>;
|
||||
|
||||
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',
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -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]));
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string>();
|
||||
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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -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']);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<IncomingFullStateConflictGateResult> {
|
||||
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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue