From e3093a416cfe806a615e9a0f45c2218c181fa15f Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Fri, 10 Jul 2026 22:11:00 +0200 Subject: [PATCH] fix(sync-core): validate remote apply results --- packages/sync-core/src/index.ts | 1 + packages/sync-core/src/ports.ts | 18 +++ packages/sync-core/src/remote-apply.ts | 148 +++++++++++------- packages/sync-core/tests/remote-apply.spec.ts | 147 ++++++++++++++++- 4 files changed, 259 insertions(+), 55 deletions(-) diff --git a/packages/sync-core/src/index.ts b/packages/sync-core/src/index.ts index 3f6ed1ab7b..835011137b 100644 --- a/packages/sync-core/src/index.ts +++ b/packages/sync-core/src/index.ts @@ -109,6 +109,7 @@ export type { ConflictUiPort, DeferredLocalActionsPort, OperationApplyPort, + ReducerCommitAwareOperationApplyPort, RemoteApplyWindowPort, SyncActionLike, } from './ports'; diff --git a/packages/sync-core/src/ports.ts b/packages/sync-core/src/ports.ts index 76396607a4..da02c6fece 100644 --- a/packages/sync-core/src/ports.ts +++ b/packages/sync-core/src/ports.ts @@ -27,6 +27,24 @@ export interface OperationApplyPort = Opera ): Promise>; } +/** + * Operation applier contract for crash-safe remote application. + * + * Unlike the general-purpose {@link OperationApplyPort}, remote application + * requires the reducer-commit callback: the coordinator must durably checkpoint + * the whole reducer batch before archive side effects can begin. + */ +export interface ReducerCommitAwareOperationApplyPort< + TOperation extends Operation = Operation, +> { + applyOperations( + ops: TOperation[], + options: ApplyOperationsOptions & { + onReducersCommitted: (ops: TOperation[]) => Promise; + }, + ): Promise>; +} + /** * Port for dispatching host actions. * diff --git a/packages/sync-core/src/remote-apply.ts b/packages/sync-core/src/remote-apply.ts index 70544f8d81..1c5dc1cf26 100644 --- a/packages/sync-core/src/remote-apply.ts +++ b/packages/sync-core/src/remote-apply.ts @@ -1,6 +1,6 @@ import type { ApplyOperationsResult } from './apply.types'; import type { Operation } from './operation.types'; -import type { OperationApplyPort } from './ports'; +import type { ReducerCommitAwareOperationApplyPort } from './ports'; export interface RemoteOperationAppendResult< TOperation extends Operation = Operation, @@ -36,7 +36,7 @@ export interface ApplyRemoteOperationsOptions< > { ops: TOperation[]; store: RemoteOperationApplyStorePort; - applier: OperationApplyPort; + applier: ReducerCommitAwareOperationApplyPort; isFullStateOperation?: (op: TOperation) => boolean; } @@ -71,14 +71,14 @@ const emptyRemoteApplyResult = < * the generic ordering: * * 1. append incoming remote ops as pending, skipping duplicates atomically; - * 2. apply only the newly appended ops through the host applier; - * 3. mark successfully applied seqs; - * 4. merge applied remote vector clocks; + * 2. bulk-apply only the newly appended ops through the host applier; + * 3. at reducer commit, mark the whole batch archive_pending and merge all + * reducer-committed vector clocks before archive side effects begin; + * 4. mark archive-complete seqs applied; * 5. after applying a full-state op, clear older full-state ops while * retaining every full-state op of this batch (incl. archive_pending); - * 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). + * 6. mark only the attempted archive failure as failed; unattempted successors + * remain archive_pending for ordered startup recovery. */ export const applyRemoteOperations = async < TOperation extends Operation = Operation, @@ -114,48 +114,62 @@ export const applyRemoteOperations = async < }); let reducerCommitCallbackCount = 0; + let reducerCommitCallbackError: Error | undefined; let reducerCommitPromise: Promise | undefined; - const applyResult = await applier.applyOperations(appendResult.writtenOps, { - onReducersCommitted: (reducerCommittedOps) => { - reducerCommitCallbackCount++; - if (reducerCommitCallbackCount !== 1) { - reducerCommitPromise = Promise.reject( - new Error( + let applyResult: ApplyOperationsResult; + try { + applyResult = await applier.applyOperations(appendResult.writtenOps, { + onReducersCommitted: (reducerCommittedOps) => { + reducerCommitCallbackCount++; + if (reducerCommitCallbackCount !== 1) { + reducerCommitCallbackError = new Error( 'applyRemoteOperations: reducer-commit callback must be invoked exactly once.', - ), - ); - return reducerCommitPromise; - } - - const isEntireAppendedBatch = - reducerCommittedOps.length === appendResult.writtenOps.length && - reducerCommittedOps.every( - (op, index) => op.id === appendResult.writtenOps[index]?.id, - ); - if (!isEntireAppendedBatch) { - reducerCommitPromise = Promise.reject( - new Error( - 'applyRemoteOperations: reducer-commit callback must contain the entire appended batch in order.', - ), - ); - return reducerCommitPromise; - } - - reducerCommitPromise = (async () => { - const reducerCommittedSeqs = reducerCommittedOps - .map((op) => opIdToSeq.get(op.id)) - .filter((seq): seq is number => seq !== undefined); - if (reducerCommittedSeqs.length !== reducerCommittedOps.length) { - throw new Error( - 'applyRemoteOperations: reducer commit contained an operation outside the appended batch.', ); + throw reducerCommitCallbackError; } - await store.markArchivePending(reducerCommittedSeqs); - await store.mergeRemoteOpClocks(reducerCommittedOps); - })(); - return reducerCommitPromise; - }, - }); + + const isEntireAppendedBatch = + reducerCommittedOps.length === appendResult.writtenOps.length && + reducerCommittedOps.every( + (op, index) => op.id === appendResult.writtenOps[index]?.id, + ); + if (!isEntireAppendedBatch) { + reducerCommitCallbackError = new Error( + 'applyRemoteOperations: reducer-commit callback must contain the entire appended batch in order.', + ); + throw reducerCommitCallbackError; + } + + reducerCommitPromise = (async () => { + const reducerCommittedSeqs = appendResult.writtenOps + .map((op) => opIdToSeq.get(op.id)) + .filter((seq): seq is number => seq !== undefined); + if (reducerCommittedSeqs.length !== appendResult.writtenOps.length) { + throw new Error( + 'applyRemoteOperations: reducer commit contained an operation outside the appended batch.', + ); + } + await store.markArchivePending(reducerCommittedSeqs); + await store.mergeRemoteOpClocks(appendResult.writtenOps); + })(); + return reducerCommitPromise; + }, + }); + } catch (applyError) { + // A valid first callback may already have started durable bookkeeping before + // a duplicate callback (or another applier error) throws. Observe that + // promise before propagating so it cannot become an unhandled rejection. + if (reducerCommitPromise) { + await reducerCommitPromise; + } + throw applyError; + } + if (reducerCommitCallbackError) { + if (reducerCommitPromise) { + await reducerCommitPromise; + } + throw reducerCommitCallbackError; + } if (reducerCommitCallbackCount !== 1 || reducerCommitPromise === undefined) { throw new Error( 'applyRemoteOperations: applier did not invoke the reducer-commit callback.', @@ -165,12 +179,40 @@ export const applyRemoteOperations = async < // awaiting its returned promise. Pending ops must never be marked applied // before the reducer/archive checkpoint is durable. await reducerCommitPromise; - if (applyResult.failedOp && !opIdToSeq.has(applyResult.failedOp.op.id)) { + + const appliedOpCount = applyResult.appliedOps.length; + const hasExactAppliedPrefix = + appliedOpCount <= appendResult.writtenOps.length && + applyResult.appliedOps.every( + (op, index) => op.id === appendResult.writtenOps[index]?.id, + ); + if (!hasExactAppliedPrefix) { throw new Error( - `applyRemoteOperations: applier reported unknown failed op ${applyResult.failedOp.op.id}.`, + 'applyRemoteOperations: applied operations must be the exact ordered prefix of the appended batch.', ); } - const appliedSeqs = applyResult.appliedOps + + const expectedFailedOp = appendResult.writtenOps[appliedOpCount]; + if ( + (applyResult.failedOp && applyResult.failedOp.op.id !== expectedFailedOp?.id) || + (!applyResult.failedOp && expectedFailedOp !== undefined) + ) { + const reportedFailedOpId = applyResult.failedOp?.op.id ?? 'none'; + const expectedFailedOpId = expectedFailedOp?.id ?? 'none'; + throw new Error( + 'applyRemoteOperations: applier must report the failed operation immediately after the applied prefix. ' + + `Reported ${reportedFailedOpId}; expected ${expectedFailedOpId}.`, + ); + } + + const authoritativeAppliedOps = appendResult.writtenOps.slice(0, appliedOpCount); + const authoritativeFailedOp = applyResult.failedOp + ? { + op: expectedFailedOp!, + error: applyResult.failedOp.error, + } + : undefined; + const appliedSeqs = authoritativeAppliedOps .map((op) => opIdToSeq.get(op.id)) .filter((seq): seq is number => seq !== undefined); @@ -179,7 +221,7 @@ export const applyRemoteOperations = async < if (appliedSeqs.length > 0) { await store.markApplied(appliedSeqs); - const appliedFullStateOpIds = applyResult.appliedOps + const appliedFullStateOpIds = authoritativeAppliedOps .filter(isFullStateOperation) .map((op) => op.id); @@ -197,7 +239,7 @@ export const applyRemoteOperations = async < } } - const failedOpIds = applyResult.failedOp ? [applyResult.failedOp.op.id] : []; + const failedOpIds = authoritativeFailedOp ? [authoritativeFailedOp.op.id] : []; if (failedOpIds.length > 0) { await store.markFailed(failedOpIds); @@ -206,10 +248,10 @@ export const applyRemoteOperations = async < return { appendedOps: appendResult.writtenOps, skippedCount: appendResult.skippedCount, - appliedOps: applyResult.appliedOps, + appliedOps: authoritativeAppliedOps, appliedSeqs, clearedFullStateOpCount, - failedOp: applyResult.failedOp, + failedOp: authoritativeFailedOp, failedOpIds, }; }; diff --git a/packages/sync-core/tests/remote-apply.spec.ts b/packages/sync-core/tests/remote-apply.spec.ts index bdfedb28db..478d9bedb3 100644 --- a/packages/sync-core/tests/remote-apply.spec.ts +++ b/packages/sync-core/tests/remote-apply.spec.ts @@ -1,7 +1,11 @@ import { describe, expect, it, vi } from 'vitest'; import { applyRemoteOperations } from '../src/remote-apply'; import type { RemoteOperationApplyStorePort } from '../src/remote-apply'; -import type { Operation, OperationApplyPort } from '../src'; +import type { + Operation, + OperationApplyPort, + ReducerCommitAwareOperationApplyPort, +} from '../src'; const createOperation = (id: string, opType = 'UPD'): Operation => ({ id, @@ -31,7 +35,7 @@ const createStore = (appendResult: { const createApplier = ( result: Awaited>['applyOperations']>>, -): OperationApplyPort> => ({ +): ReducerCommitAwareOperationApplyPort> => ({ applyOperations: vi.fn(async (ops, options) => { await options?.onReducersCommitted?.(ops); return result; @@ -238,6 +242,145 @@ describe('applyRemoteOperations', () => { ); expect(store.markApplied).not.toHaveBeenCalled(); }); + + it('uses authoritative written operations for clocks, cleanup, and results', async () => { + const writtenImport = createOperation('sync-import-1', 'SYNC_IMPORT'); + writtenImport.vectorClock = { authoritative: 7 }; + const callbackClone = { + ...writtenImport, + vectorClock: { forgedCallback: 99 }, + }; + const appliedClone = { + ...writtenImport, + opType: 'UPD', + vectorClock: { forgedResult: 100 }, + }; + const store = createStore({ + seqs: [11], + writtenOps: [writtenImport], + skippedCount: 0, + }); + const applier: OperationApplyPort> = { + applyOperations: vi.fn(async (_ops, options) => { + await options?.onReducersCommitted?.([callbackClone]); + return { appliedOps: [appliedClone] }; + }), + }; + + const result = await applyRemoteOperations({ + ops: [writtenImport], + store, + applier, + isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT', + }); + + expect(store.mergeRemoteOpClocks).toHaveBeenCalledWith([writtenImport]); + expect(store.clearFullStateOpsExcept).toHaveBeenCalledWith(['sync-import-1']); + expect(result.appendedOps[0]).toBe(writtenImport); + expect(result.appliedOps[0]).toBe(writtenImport); + }); + + it('rejects applied operations that are not the exact ordered written prefix', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const store = createStore({ + seqs: [1, 2], + writtenOps: [op1, op2], + skippedCount: 0, + }); + const applier = createApplier({ appliedOps: [op2, op1] }); + + await expect( + applyRemoteOperations({ ops: [op1, op2], store, applier }), + ).rejects.toThrow('exact ordered prefix'); + expect(store.markApplied).not.toHaveBeenCalled(); + }); + + it('rejects a partial applied prefix without the next failed operation', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const store = createStore({ + seqs: [1, 2], + writtenOps: [op1, op2], + skippedCount: 0, + }); + const applier = createApplier({ appliedOps: [op1] }); + + await expect( + applyRemoteOperations({ ops: [op1, op2], store, applier }), + ).rejects.toThrow('failed operation immediately after the applied prefix'); + expect(store.markApplied).not.toHaveBeenCalled(); + }); + + it('rejects a failed operation that is not immediately after the applied prefix', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const op3 = createOperation('op-3'); + const store = createStore({ + seqs: [1, 2, 3], + writtenOps: [op1, op2, op3], + skippedCount: 0, + }); + const applier = createApplier({ + appliedOps: [op1], + failedOp: { op: op3, error: new Error('wrong failure') }, + }); + + await expect( + applyRemoteOperations({ ops: [op1, op2, op3], store, applier }), + ).rejects.toThrow('failed operation immediately after the applied prefix'); + expect(store.markFailed).not.toHaveBeenCalled(); + }); + + it('throws synchronously from a malformed reducer-commit callback', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const store = createStore({ + seqs: [1, 2], + writtenOps: [op1, op2], + skippedCount: 0, + }); + let callbackThrewSynchronously = false; + const applier: OperationApplyPort> = { + applyOperations: vi.fn(async (_ops, options) => { + try { + void options?.onReducersCommitted?.([op2, op1]); + } catch (error) { + callbackThrewSynchronously = true; + throw error; + } + return { appliedOps: [op1, op2] }; + }), + }; + + await expect( + applyRemoteOperations({ ops: [op1, op2], store, applier }), + ).rejects.toThrow('entire appended batch in order'); + expect(callbackThrewSynchronously).toBe(true); + }); + + it('throws synchronously from a duplicate reducer-commit callback', async () => { + const op = createOperation('op-1'); + const store = createStore({ seqs: [1], writtenOps: [op], skippedCount: 0 }); + let duplicateCallbackThrewSynchronously = false; + const applier: OperationApplyPort> = { + applyOperations: vi.fn(async (ops, options) => { + void options?.onReducersCommitted?.(ops); + try { + void options?.onReducersCommitted?.(ops); + } catch (error) { + duplicateCallbackThrewSynchronously = true; + throw error; + } + return { appliedOps: ops }; + }), + }; + + await expect(applyRemoteOperations({ ops: [op], store, applier })).rejects.toThrow( + 'exactly once', + ); + expect(duplicateCallbackThrewSynchronously).toBe(true); + }); }); const jasmineLikeObjectContainingFunction = (): object =>