fix(sync): enforce upload completion contracts

packages/sync-core/src/remote-apply.ts

@@ -113,20 +113,58 @@ export const applyRemoteOperations = async <
     }
   });
 
+  let reducerCommitCallbackCount = 0;
+  let reducerCommitPromise: Promise<void> | undefined;
   const applyResult = await applier.applyOperations(appendResult.writtenOps, {
-    onReducersCommitted: async (reducerCommittedOps) => {
-      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.',
+    onReducersCommitted: (reducerCommittedOps) => {
+      reducerCommitCallbackCount++;
+      if (reducerCommitCallbackCount !== 1) {
+        reducerCommitPromise = Promise.reject(
+          new Error(
+            'applyRemoteOperations: reducer-commit callback must be invoked exactly once.',
+          ),
         );
+        return reducerCommitPromise;
       }
-      await store.markArchivePending(reducerCommittedSeqs);
-      await store.mergeRemoteOpClocks(reducerCommittedOps);
+
+      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.',
+          );
+        }
+        await store.markArchivePending(reducerCommittedSeqs);
+        await store.mergeRemoteOpClocks(reducerCommittedOps);
+      })();
+      return reducerCommitPromise;
     },
   });
+  if (reducerCommitCallbackCount !== 1 || reducerCommitPromise === undefined) {
+    throw new Error(
+      'applyRemoteOperations: applier did not invoke the reducer-commit callback.',
+    );
+  }
+  // Also await host bookkeeping when an applier invoked the callback without
+  // 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)) {
     throw new Error(
       `applyRemoteOperations: applier reported unknown failed op ${applyResult.failedOp.op.id}.`,

packages/sync-core/tests/remote-apply.spec.ts

@@ -187,6 +187,57 @@ describe('applyRemoteOperations', () => {
     );
     expect(store.markFailed).not.toHaveBeenCalled();
   });
+
+  it('fails closed when the applier ignores the reducer-commit callback', async () => {
+    const op = createOperation('op-1');
+    const store = createStore({ seqs: [1], writtenOps: [op], skippedCount: 0 });
+    const applier: OperationApplyPort<Operation<string>> = {
+      applyOperations: vi.fn().mockResolvedValue({ appliedOps: [op] }),
+    };
+
+    await expect(applyRemoteOperations({ ops: [op], store, applier })).rejects.toThrow(
+      'reducer-commit callback',
+    );
+    expect(store.markApplied).not.toHaveBeenCalled();
+  });
+
+  it('fails closed when the reducer-commit callback omits an appended op', async () => {
+    const op1 = createOperation('op-1');
+    const op2 = createOperation('op-2');
+    const store = createStore({
+      seqs: [1, 2],
+      writtenOps: [op1, op2],
+      skippedCount: 0,
+    });
+    const applier: OperationApplyPort<Operation<string>> = {
+      applyOperations: vi.fn(async (_ops, options) => {
+        await options?.onReducersCommitted?.([op1]);
+        return { appliedOps: [op1, op2] };
+      }),
+    };
+
+    await expect(
+      applyRemoteOperations({ ops: [op1, op2], store, applier }),
+    ).rejects.toThrow('entire appended batch');
+    expect(store.markApplied).not.toHaveBeenCalled();
+  });
+
+  it('fails closed when the reducer-commit callback is invoked twice', async () => {
+    const op = createOperation('op-1');
+    const store = createStore({ seqs: [1], writtenOps: [op], skippedCount: 0 });
+    const applier: OperationApplyPort<Operation<string>> = {
+      applyOperations: vi.fn(async (ops, options) => {
+        await options?.onReducersCommitted?.(ops);
+        await options?.onReducersCommitted?.(ops);
+        return { appliedOps: ops };
+      }),
+    };
+
+    await expect(applyRemoteOperations({ ops: [op], store, applier })).rejects.toThrow(
+      'exactly once',
+    );
+    expect(store.markApplied).not.toHaveBeenCalled();
+  });
 });
 
 const jasmineLikeObjectContainingFunction = (): object =>