fix(sync-core): validate remote apply results

packages/sync-core/src/index.ts

@@ -109,6 +109,7 @@ export type {
   ConflictUiPort,
   DeferredLocalActionsPort,
   OperationApplyPort,
+  ReducerCommitAwareOperationApplyPort,
   RemoteApplyWindowPort,
   SyncActionLike,
 } from './ports';

packages/sync-core/src/ports.ts

@@ -27,6 +27,24 @@ export interface OperationApplyPort<TOperation extends Operation<string> = Opera
   ): Promise<ApplyOperationsResult<TOperation>>;
 }
 
+/**
+ * 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<string> = Operation,
+> {
+  applyOperations(
+    ops: TOperation[],
+    options: ApplyOperationsOptions & {
+      onReducersCommitted: (ops: TOperation[]) => Promise<void>;
+    },
+  ): Promise<ApplyOperationsResult<TOperation>>;
+}
+
 /**
  * Port for dispatching host actions.
  *

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<string> = Operation,
@@ -36,7 +36,7 @@ export interface ApplyRemoteOperationsOptions<
 > {
   ops: TOperation[];
   store: RemoteOperationApplyStorePort<TOperation>;
-  applier: OperationApplyPort<TOperation>;
+  applier: ReducerCommitAwareOperationApplyPort<TOperation>;
   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<string> = Operation,
@@ -114,48 +114,62 @@ export const applyRemoteOperations = async <
   });
 
   let reducerCommitCallbackCount = 0;
+  let reducerCommitCallbackError: Error | undefined;
   let reducerCommitPromise: Promise<void> | undefined;
-  const applyResult = await applier.applyOperations(appendResult.writtenOps, {
-    onReducersCommitted: (reducerCommittedOps) => {
-      reducerCommitCallbackCount++;
-      if (reducerCommitCallbackCount !== 1) {
-        reducerCommitPromise = Promise.reject(
-          new Error(
+  let applyResult: ApplyOperationsResult<TOperation>;
+  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,
   };
 };

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<string> => ({
   id,
@@ -31,7 +35,7 @@ const createStore = (appendResult: {
 
 const createApplier = (
   result: Awaited<ReturnType<OperationApplyPort<Operation<string>>['applyOperations']>>,
-): OperationApplyPort<Operation<string>> => ({
+): ReducerCommitAwareOperationApplyPort<Operation<string>> => ({
   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<Operation<string>> = {
+      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<Operation<string>> = {
+      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<Operation<string>> = {
+      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 =>