Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-07-17 16:37:43 +00:00
Code Issues Projects Releases Wiki Activity

fix(sync): harden op-log replay and recovery (#8978)

Browse source 
* fix(sync): harden operation-log failure recovery

Keep compaction, archive replacement, legacy recovery, and reducer replay checkpointing deterministic across crashes and concurrent clients.

Refs #8958

* fix(sync): harden remote operation recovery

Persist reducer failures across hydration and fail closed for full-state operations. Serialize recovery and archive mutations, and report skipped emergency compaction accurately.

* fix(sync): preserve operations across replay failures
This commit is contained in:
Johannes Millan 2026-07-13 20:03:42 +02:00 • committed by GitHub
parent aa75e6f7ee
commit 5624f6891d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
46 changed files with 2839 additions and 554 deletions
Download patch file Download diff file Expand all files Collapse all files

16
packages/sync-core/src/apply.types.ts
View file

@ -1,5 +1,10 @@
import type { Operation } from './operation.types';
export interface OperationApplyFailure<TOperation extends Operation<string> = Operation> {
op: TOperation;
error: Error;
}
/**
* Result of applying a batch of operations.
*
@ -10,13 +15,14 @@ export interface ApplyOperationsResult<TOperation extends Operation<string> = Op
/** Operations that were successfully applied. */
appliedOps: TOperation[];
/** Operations skipped because their host reducer/conversion threw. */
reducerFailures?: OperationApplyFailure<TOperation>[];
/**
* If an error occurred, this contains the failed operation and the error.
* 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.
* The failed op and the successfully reducer-applied operations after it did
* NOT complete their archive side effects. Their reducer effects DID commit,
* so retry paths must use `skipReducerDispatch` to avoid double-applying them.
*/
failedOp?: {
op: TOperation;

6
packages/sync-core/src/index.ts
View file

@ -75,7 +75,11 @@ export { createLwwUpdateActionTypeHelpers } from './lww-update-action-types';
export type { LwwUpdateActionTypeHelpers } from './lww-update-action-types';
// Apply-operation result and option types.
export type { ApplyOperationsResult, ApplyOperationsOptions } from './apply.types';
export type {
ApplyOperationsResult,
ApplyOperationsOptions,
OperationApplyFailure,
} from './apply.types';
// Generic operation replay coordinator.
export { replayOperationBatch } from './replay-coordinator';

8
packages/sync-core/src/operation.types.ts
View file

@ -147,6 +147,14 @@ export interface OperationLogEntry<TOperation extends Operation<string> = Operat
*/
rejectedAt?: number;
/**
* Timestamp (epoch ms) when replay skipped this operation because conversion,
* schema migration, or reducer application could not produce state. Unlike
* ordinary conflict rejection, this operation must never be replayed during
* hydration.
*/
reducerRejectedAt?: number;
/**
* For remote ops only: tracks whether the op was successfully applied to
* local state. `archive_pending` means reducers and clocks committed but

16
packages/sync-core/src/ports.ts
View file

@ -1,4 +1,8 @@
import type { ApplyOperationsOptions, ApplyOperationsResult } from './apply.types';
import type {
ApplyOperationsOptions,
ApplyOperationsResult,
OperationApplyFailure,
} from './apply.types';
import type { Operation } from './operation.types';
export type SyncPortMeta = Record<string, string | number | boolean | null | undefined>;
@ -22,7 +26,10 @@ export interface OperationApplyPort<TOperation extends Operation<string> = Opera
ops: TOperation[],
options?: ApplyOperationsOptions & {
/** Persist reducer-commit bookkeeping before post-dispatch side effects start. */
onReducersCommitted?: (ops: TOperation[]) => Promise<void>;
onReducersCommitted?: (
ops: TOperation[],
failures?: OperationApplyFailure<TOperation>[],
) => Promise<void>;
},
): Promise<ApplyOperationsResult<TOperation>>;
}
@ -40,7 +47,10 @@ export interface ReducerCommitAwareOperationApplyPort<
applyOperations(
ops: TOperation[],
options: ApplyOperationsOptions & {
onReducersCommitted: (ops: TOperation[]) => Promise<void>;
onReducersCommitted: (
ops: TOperation[],
failures?: OperationApplyFailure<TOperation>[],
) => Promise<void>;
},
): Promise<ApplyOperationsResult<TOperation>>;
}

148
packages/sync-core/src/remote-apply.ts
View file

@ -1,4 +1,4 @@
import type { ApplyOperationsResult } from './apply.types';
import type { ApplyOperationsResult, OperationApplyFailure } from './apply.types';
import type { Operation } from './operation.types';
import type { ReducerCommitAwareOperationApplyPort } from './ports';
@ -25,7 +25,11 @@ export interface RemoteOperationApplyStorePort<
options: { pendingApply: true },
): Promise<RemoteOperationAppendResult<TOperation>>;
mergeRemoteOpClocks(ops: TOperation[]): Promise<void>;
markReducersCommittedAndMergeClocks(seqs: number[], ops: TOperation[]): Promise<void>;
markReducersCommittedAndMergeClocks(
seqs: number[],
ops: TOperation[],
rejectedOpIds?: string[],
): Promise<void>;
markApplied(seqs: number[]): Promise<void>;
markFailed(opIds: string[]): Promise<void>;
clearFullStateOpsExcept(excludeIds: string[]): Promise<number>;
@ -52,6 +56,8 @@ export interface RemoteApplyOperationsResult<
clearedFullStateOpCount: number;
failedOp?: ApplyOperationsResult<TOperation>['failedOp'];
failedOpIds: string[];
reducerFailures?: OperationApplyFailure<TOperation>[];
reducerFailedOpIds?: string[];
}
const emptyRemoteApplyResult = <
@ -76,9 +82,9 @@ const emptyRemoteApplyResult = <
* 2. durably merge all newly appended remote clocks before reducers or a
* deferred-local-action window can start;
* 3. bulk-apply only the newly appended ops through the host applier;
* 4. at reducer commit, atomically checkpoint the whole batch as
* `archive_pending` and merge all reducer-committed vector clocks before
* archive side effects begin;
* 4. at reducer commit, atomically checkpoint reducer-successful ops as
* `archive_pending`, reject reducer-failed ops, and merge successful clocks
* before archive side effects begin;
* 5. mark archive-complete seqs applied;
* 6. after applying a full-state op, clear older full-state ops while
* retaining every full-state op of this batch (incl. quarantined ones);
@ -129,6 +135,8 @@ export const applyRemoteOperations = async <
let reducerCommitCallbackCount = 0;
let reducerCommitCallbackError: Error | undefined;
let reducerCommitPromise: Promise<void> | undefined;
let reducerCommittedOps: TOperation[] | undefined;
let reducerFailures: OperationApplyFailure<TOperation>[] | undefined;
let applyResult: ApplyOperationsResult<TOperation>;
const observeReducerCommitPromisePreservingPrimaryError = async (): Promise<void> => {
const promise = reducerCommitPromise;
@ -145,7 +153,7 @@ export const applyRemoteOperations = async <
};
try {
applyResult = await applier.applyOperations(appendResult.writtenOps, {
onReducersCommitted: (reducerCommittedOps) => {
onReducersCommitted: (reportedCommittedOps, reportedFailures = []) => {
reducerCommitCallbackCount++;
if (reducerCommitCallbackCount !== 1) {
reducerCommitCallbackError = new Error(
@ -154,31 +162,63 @@ export const applyRemoteOperations = async <
throw reducerCommitCallbackError;
}
const isEntireAppendedBatch =
reducerCommittedOps.length === appendResult.writtenOps.length &&
reducerCommittedOps.every(
(op, index) => op.id === appendResult.writtenOps[index]?.id,
);
if (!isEntireAppendedBatch) {
const authoritativeReducerFailures = getAuthoritativeReducerFailures(
appendResult.writtenOps,
reportedFailures,
);
const failedFullStateOp = authoritativeReducerFailures.find((failure) =>
isFullStateOperation(failure.op),
);
if (failedFullStateOp) {
reducerCommitCallbackError = new Error(
'applyRemoteOperations: reducer-commit callback must contain the entire appended batch in order.',
`applyRemoteOperations: full-state operation ${failedFullStateOp.op.id} failed during reducer replay.`,
);
throw reducerCommitCallbackError;
}
reducerFailures = authoritativeReducerFailures;
const reducerFailedIds = new Set(
authoritativeReducerFailures.map((failure) => failure.op.id),
);
const expectedCommittedOps = appendResult.writtenOps.filter(
(op) => !reducerFailedIds.has(op.id),
);
const isExactSuccessfulSubset =
reportedCommittedOps.length === expectedCommittedOps.length &&
reportedCommittedOps.every(
(op, index) => op.id === expectedCommittedOps[index]?.id,
);
if (!isExactSuccessfulSubset) {
reducerCommitCallbackError = new Error(
'applyRemoteOperations: reducer-commit callback and failures must partition the entire appended batch in order.',
);
throw reducerCommitCallbackError;
}
reducerCommittedOps = expectedCommittedOps;
reducerCommitPromise = (async () => {
const reducerCommittedSeqs = appendResult.writtenOps
const reducerCommittedSeqs = expectedCommittedOps
.map((op) => opIdToSeq.get(op.id))
.filter((seq): seq is number => seq !== undefined);
if (reducerCommittedSeqs.length !== appendResult.writtenOps.length) {
if (reducerCommittedSeqs.length !== expectedCommittedOps.length) {
throw new Error(
'applyRemoteOperations: reducer commit contained an operation outside the appended batch.',
);
}
await store.markReducersCommittedAndMergeClocks(
reducerCommittedSeqs,
appendResult.writtenOps,
const reducerFailedOpIds = authoritativeReducerFailures.map(
(failure) => failure.op.id,
);
if (reducerFailedOpIds.length > 0) {
await store.markReducersCommittedAndMergeClocks(
reducerCommittedSeqs,
expectedCommittedOps,
reducerFailedOpIds,
);
} else {
await store.markReducersCommittedAndMergeClocks(
reducerCommittedSeqs,
expectedCommittedOps,
);
}
})();
return reducerCommitPromise;
},
@ -194,7 +234,12 @@ export const applyRemoteOperations = async <
await observeReducerCommitPromisePreservingPrimaryError();
throw reducerCommitCallbackError;
}
if (reducerCommitCallbackCount !== 1 || reducerCommitPromise === undefined) {
if (
reducerCommitCallbackCount !== 1 ||
reducerCommitPromise === undefined ||
reducerCommittedOps === undefined ||
reducerFailures === undefined
) {
throw new Error(
'applyRemoteOperations: applier did not invoke the reducer-commit callback.',
);
@ -203,20 +248,37 @@ export const applyRemoteOperations = async <
// awaiting its returned promise. Pending ops must never be marked applied
// before the reducer/archive checkpoint is durable.
await reducerCommitPromise;
const authoritativeReducerCommittedOps = reducerCommittedOps;
const authoritativeReducerFailures = reducerFailures;
const appliedOpCount = applyResult.appliedOps.length;
const hasExactAppliedPrefix =
appliedOpCount <= appendResult.writtenOps.length &&
applyResult.appliedOps.every(
(op, index) => op.id === appendResult.writtenOps[index]?.id,
const resultReducerFailures = getAuthoritativeReducerFailures(
appendResult.writtenOps,
applyResult.reducerFailures ?? [],
);
const hasMatchingReducerFailures =
resultReducerFailures.length === authoritativeReducerFailures.length &&
resultReducerFailures.every(
(failure, index) => failure.op.id === authoritativeReducerFailures[index]?.op.id,
);
if (!hasExactAppliedPrefix) {
if (!hasMatchingReducerFailures) {
throw new Error(
'applyRemoteOperations: applied operations must be the exact ordered prefix of the appended batch.',
'applyRemoteOperations: applier result must report the same reducer failures as the reducer-commit callback.',
);
}
const expectedFailedOp = appendResult.writtenOps[appliedOpCount];
const appliedOpCount = applyResult.appliedOps.length;
const hasExactAppliedPrefix =
appliedOpCount <= authoritativeReducerCommittedOps.length &&
applyResult.appliedOps.every(
(op, index) => op.id === authoritativeReducerCommittedOps[index]?.id,
);
if (!hasExactAppliedPrefix) {
throw new Error(
'applyRemoteOperations: applied operations must be the exact ordered prefix of the reducer-committed operations.',
);
}
const expectedFailedOp = authoritativeReducerCommittedOps[appliedOpCount];
if (
(applyResult.failedOp && applyResult.failedOp.op.id !== expectedFailedOp?.id) ||
(!applyResult.failedOp && expectedFailedOp !== undefined)
@ -229,7 +291,10 @@ export const applyRemoteOperations = async <
);
}
const authoritativeAppliedOps = appendResult.writtenOps.slice(0, appliedOpCount);
const authoritativeAppliedOps = authoritativeReducerCommittedOps.slice(
0,
appliedOpCount,
);
const authoritativeFailedOp = applyResult.failedOp
? {
op: expectedFailedOp!,
@ -264,6 +329,7 @@ export const applyRemoteOperations = async <
}
const failedOpIds = authoritativeFailedOp ? [authoritativeFailedOp.op.id] : [];
const reducerFailedOpIds = authoritativeReducerFailures.map((failure) => failure.op.id);
if (failedOpIds.length > 0) {
await store.markFailed(failedOpIds);
@ -277,5 +343,31 @@ export const applyRemoteOperations = async <
clearedFullStateOpCount,
failedOp: authoritativeFailedOp,
failedOpIds,
...(authoritativeReducerFailures.length > 0
? { reducerFailures: authoritativeReducerFailures, reducerFailedOpIds }
: {}),
};
};
const getAuthoritativeReducerFailures = <TOperation extends Operation<string>>(
writtenOps: TOperation[],
reportedFailures: OperationApplyFailure<TOperation>[],
): OperationApplyFailure<TOperation>[] => {
const writtenOpById = new Map(writtenOps.map((op) => [op.id, op]));
const seenIds = new Set<string>();
return reportedFailures.map((failure) => {
const authoritativeOp = writtenOpById.get(failure.op.id);
if (!authoritativeOp || seenIds.has(failure.op.id)) {
throw new Error(
`applyRemoteOperations: invalid reducer failure for operation ${failure.op.id}.`,
);
}
seenIds.add(failure.op.id);
return {
op: authoritativeOp,
error:
failure.error instanceof Error ? failure.error : new Error(String(failure.error)),
};
});
};

60
packages/sync-core/src/replay-coordinator.ts
View file

@ -1,4 +1,8 @@
import type { ApplyOperationsOptions, ApplyOperationsResult } from './apply.types';
import type {
ApplyOperationsOptions,
ApplyOperationsResult,
OperationApplyFailure,
} from './apply.types';
import type { Operation } from './operation.types';
import type {
ActionDispatchPort,
@ -29,13 +33,18 @@ export interface OperationReplayCoordinatorOptions<
applyOptions?: ApplyOperationsOptions;
dispatcher: ActionDispatchPort<TBulkAction>;
createBulkApplyAction: (ops: TOperation[]) => TBulkAction;
/** Reads reducer failures captured synchronously during the bulk dispatch. */
getReducerFailures?: () => OperationApplyFailure<TOperation>[];
remoteApplyWindow: RemoteApplyWindowPort;
deferredLocalActions: DeferredLocalActionsPort;
archiveSideEffects?: ArchiveSideEffectPort<TReplayAction>;
operationToAction?: (op: TOperation) => TReplayAction;
isArchiveAffectingAction?: (action: TReplayAction) => boolean;
onRemoteArchiveDataApplied?: () => void;
onReducersCommitted?: (ops: TOperation[]) => Promise<void>;
onReducersCommitted?: (
ops: TOperation[],
failures: OperationApplyFailure<TOperation>[],
) => Promise<void>;
onArchiveSideEffectError?: (
context: OperationReplayArchiveFailureContext<TOperation>,
) => void;
@ -75,8 +84,8 @@ export const yieldToEventLoop = (): Promise<void> =>
* 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;
* 3. yield once, then exclude any host-reported reducer failures;
* 4. checkpoint and run archive side effects only for reducer-successful ops;
* 5. start post-sync cooldown before closing the remote-apply window;
* 6. close the window and flush deferred local actions.
*/
@ -89,6 +98,7 @@ export const replayOperationBatch = async <
applyOptions = {},
dispatcher,
createBulkApplyAction,
getReducerFailures,
remoteApplyWindow,
deferredLocalActions,
archiveSideEffects,
@ -116,16 +126,21 @@ export const replayOperationBatch = async <
}
remoteApplyWindow.startApplyingRemoteOps();
let reducerCommittedOps = ops;
let reducerFailures: OperationApplyFailure<TOperation>[] = [];
try {
if (!skipReducerDispatch) {
dispatcher.dispatch(createBulkApplyAction(ops));
await waitForEventLoop();
await onReducersCommitted?.(ops);
reducerFailures = validateReducerFailures(ops, getReducerFailures?.() ?? []);
const reducerFailedIds = new Set(reducerFailures.map((failure) => failure.op.id));
reducerCommittedOps = ops.filter((op) => !reducerFailedIds.has(op.id));
await onReducersCommitted?.(reducerCommittedOps, reducerFailures);
}
if (!isLocalHydration && archiveSideEffects !== undefined && operationToAction) {
const archiveResult = await processArchiveSideEffects({
ops,
ops: reducerCommittedOps,
archiveSideEffects,
operationToAction,
isArchiveAffectingAction,
@ -137,6 +152,7 @@ export const replayOperationBatch = async <
return {
appliedOps: archiveResult.appliedOps,
failedOp: archiveResult.failedOp,
...(reducerFailures.length > 0 ? { reducerFailures } : {}),
};
}
@ -157,7 +173,37 @@ export const replayOperationBatch = async <
await deferredLocalActions.processDeferredActions();
}
return { appliedOps: ops };
return {
appliedOps: reducerCommittedOps,
...(reducerFailures.length > 0 ? { reducerFailures } : {}),
};
};
const validateReducerFailures = <TOperation extends Operation<string>>(
ops: TOperation[],
reportedFailures: OperationApplyFailure<TOperation>[],
): OperationApplyFailure<TOperation>[] => {
const opById = new Map(ops.map((op) => [op.id, op]));
const seenIds = new Set<string>();
return reportedFailures.map((failure) => {
const authoritativeOp = opById.get(failure.op.id);
if (!authoritativeOp) {
throw new Error(
`replayOperationBatch: reducer failure references unknown operation ${failure.op.id}.`,
);
}
if (seenIds.has(failure.op.id)) {
throw new Error(
`replayOperationBatch: reducer failure reported more than once for ${failure.op.id}.`,
);
}
seenIds.add(failure.op.id);
return {
op: authoritativeOp,
error: toError(failure.error),
};
});
};
const processArchiveSideEffects = async <

72
packages/sync-core/tests/remote-apply.spec.ts
View file

@ -180,6 +180,78 @@ describe('applyRemoteOperations', () => {
expect(result.failedOpIds).toEqual(['op-2']);
});
it('rejects reducer-failed ops while checkpointing and applying successful successors', async () => {
const op1 = createOperation('op-1');
const op2 = createOperation('op-2');
const op3 = createOperation('op-3');
const reducerError = new Error('reducer failure');
const store = createStore({
seqs: [1, 2, 3],
writtenOps: [op1, op2, op3],
skippedCount: 0,
});
const applier: ReducerCommitAwareOperationApplyPort<Operation<string>> = {
applyOperations: vi.fn(async (_ops, options) => {
await options.onReducersCommitted([op1, op3], [{ op: op2, error: reducerError }]);
return {
appliedOps: [op1, op3],
reducerFailures: [{ op: op2, error: reducerError }],
};
}),
};
const result = await applyRemoteOperations({
ops: [op1, op2, op3],
store,
applier,
});
expect(store.markReducersCommittedAndMergeClocks).toHaveBeenCalledWith(
[1, 3],
[op1, op3],
['op-2'],
);
expect(store.markApplied).toHaveBeenCalledWith([1, 3]);
expect(store.markFailed).not.toHaveBeenCalled();
expect(result.appliedOps).toEqual([op1, op3]);
expect(result.reducerFailedOpIds).toEqual(['op-2']);
expect(result.reducerFailures).toEqual([{ op: op2, error: reducerError }]);
});
it('fails closed before checkpointing when a full-state reducer fails', async () => {
const fullStateOp = {
...createOperation('sync-import'),
opType: 'SYNC_IMPORT',
};
const reducerError = new Error('full-state reducer failure');
const store = createStore({
seqs: [1],
writtenOps: [fullStateOp],
skippedCount: 0,
});
const applier: ReducerCommitAwareOperationApplyPort<Operation<string>> = {
applyOperations: vi.fn(async (_ops, options) => {
await options.onReducersCommitted([], [{ op: fullStateOp, error: reducerError }]);
return {
appliedOps: [],
reducerFailures: [{ op: fullStateOp, error: reducerError }],
};
}),
};
await expect(
applyRemoteOperations({
ops: [fullStateOp],
store,
applier,
isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT',
}),
).rejects.toThrow(/full-state.*reducer/i);
expect(store.markReducersCommittedAndMergeClocks).not.toHaveBeenCalled();
expect(store.markApplied).not.toHaveBeenCalled();
});
it('rejects an applier result whose failed op is not in the appended batch', async () => {
const op1 = createOperation('op-1');
const unknownFailedOp = createOperation('unknown-failed-op');

46
packages/sync-core/tests/replay-coordinator.spec.ts
View file

@ -269,6 +269,52 @@ describe('replayOperationBatch', () => {
]);
});
it('skips reducer-failed operations during checkpointing and archive handling', async () => {
const op1 = createOperation('op-1');
const op2 = createOperation('op-2');
const op3 = createOperation('op-3');
const reducerError = new Error('reducer failed');
const onReducersCommitted = vi.fn(async () => undefined);
const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
handleOperation: vi.fn(async () => undefined),
};
const result = await replayOperationBatch({
ops: [op1, op2, op3],
dispatcher: { dispatch: vi.fn() },
createBulkApplyAction: (operations) => ({
type: '[Test] Bulk Apply',
operations,
}),
getReducerFailures: () => [{ op: op2, error: reducerError }],
remoteApplyWindow: createRemoteApplyWindow([]),
deferredLocalActions: createDeferredLocalActions([]),
archiveSideEffects,
operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }),
onReducersCommitted,
yieldToEventLoop: vi.fn(async () => undefined),
});
expect(onReducersCommitted).toHaveBeenCalledOnce();
expect(onReducersCommitted).toHaveBeenCalledWith(
[op1, op3],
[{ op: op2, error: reducerError }],
);
expect(archiveSideEffects.handleOperation).toHaveBeenCalledTimes(2);
expect(archiveSideEffects.handleOperation).toHaveBeenNthCalledWith(
1,
expect.objectContaining({ opId: 'op-1' }),
);
expect(archiveSideEffects.handleOperation).toHaveBeenNthCalledWith(
2,
expect.objectContaining({ opId: 'op-3' }),
);
expect(result).toEqual({
appliedOps: [op1, op3],
reducerFailures: [{ op: op2, error: reducerError }],
});
});
it('runs only archive side effects when skipReducerDispatch is set (retry path)', async () => {
const callOrder: string[] = [];
const ops = [createOperation('op-1'), createOperation('op-2')];