From 0939f5af69a859884f5c7dbf7ddbc3d9ceaa126f Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Fri, 10 Jul 2026 19:35:37 +0200 Subject: [PATCH] fix(sync): enforce upload completion contracts --- .../src/supersync-http-contract.ts | 10 +++- .../tests/supersync-http-contract.spec.ts | 15 +++-- .../tests/sync-compressed-body.routes.spec.ts | 43 ++++++++++++++ packages/sync-core/src/remote-apply.ts | 56 ++++++++++++++++--- packages/sync-core/tests/remote-apply.spec.ts | 51 +++++++++++++++++ .../sync/immediate-upload.service.spec.ts | 43 ++++++++++++++ .../op-log/sync/immediate-upload.service.ts | 41 ++++++++++++-- 7 files changed, 239 insertions(+), 20 deletions(-) diff --git a/packages/shared-schema/src/supersync-http-contract.ts b/packages/shared-schema/src/supersync-http-contract.ts index b128400cbf..f510c270f7 100644 --- a/packages/shared-schema/src/supersync-http-contract.ts +++ b/packages/shared-schema/src/supersync-http-contract.ts @@ -74,8 +74,16 @@ export const SuperSyncOperationSchema = z.object({ syncImportReason: z.enum(SUPER_SYNC_IMPORT_REASONS).optional(), }); +// Upload requests are envelopes for independently validated operations. Keep +// the transport shape numeric, but let the server return a per-op +// INVALID_SCHEMA_VERSION result so one malformed op cannot reject and stall +// every valid sibling in the batch. Download/response schemas remain strict. +const SuperSyncUploadOperationSchema = SuperSyncOperationSchema.extend({ + schemaVersion: z.number(), +}); + export const SuperSyncUploadOpsRequestSchema = z.object({ - ops: z.array(SuperSyncOperationSchema).min(1).max(SUPER_SYNC_MAX_OPS_PER_UPLOAD), + ops: z.array(SuperSyncUploadOperationSchema).min(1).max(SUPER_SYNC_MAX_OPS_PER_UPLOAD), clientId: SuperSyncClientIdSchema, lastKnownServerSeq: z.number().optional(), requestId: SuperSyncRequestIdSchema.optional(), diff --git a/packages/shared-schema/tests/supersync-http-contract.spec.ts b/packages/shared-schema/tests/supersync-http-contract.spec.ts index c0f725540b..10e8d30f4c 100644 --- a/packages/shared-schema/tests/supersync-http-contract.spec.ts +++ b/packages/shared-schema/tests/supersync-http-contract.spec.ts @@ -4,6 +4,7 @@ import { SUPER_SYNC_MAX_OPS_PER_UPLOAD, SuperSyncDownloadOpsQuerySchema, SuperSyncDownloadOpsResponseSchema, + SuperSyncOperationSchema, SuperSyncUploadOpsRequestSchema, SuperSyncUploadSnapshotRequestSchema, } from '../src/supersync-http-contract'; @@ -73,12 +74,18 @@ describe('SuperSync HTTP contract schemas', () => { }); it.each([0, 1.5, -1, 101])( - 'rejects malformed operation schema version %s', + 'passes operation schema version %s through the upload transport schema', (schemaVersion) => { + const parsed = SuperSyncUploadOpsRequestSchema.parse({ + ops: [{ ...createValidOperation(), schemaVersion }], + clientId: 'client_1', + }); + + expect(parsed.ops[0].schemaVersion).toBe(schemaVersion); expect(() => - SuperSyncUploadOpsRequestSchema.parse({ - ops: [{ ...createValidOperation(), schemaVersion }], - clientId: 'client_1', + SuperSyncOperationSchema.parse({ + ...createValidOperation(), + schemaVersion, }), ).toThrow(); }, diff --git a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts index 0acbb475f0..d9cd53a1c7 100644 --- a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts +++ b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts @@ -212,6 +212,49 @@ describe('Sync compressed body routes', () => { expect(quotaCall[1]).toBeLessThan(payloadSize); }); + it('should pass mixed schema versions to per-operation validation', async () => { + const clientId = 'mixed-schema-client'; + const validOp = createOp(clientId); + const invalidOp = { + ...createOp(clientId), + id: 'invalid-schema-op', + schemaVersion: 101, + }; + mocks.syncService.uploadOps.mockResolvedValueOnce([ + { opId: validOp.id, accepted: true, serverSeq: 1 }, + { + opId: invalidOp.id, + accepted: false, + error: 'Invalid schema version: 101', + errorCode: SYNC_ERROR_CODES.INVALID_SCHEMA_VERSION, + }, + ]); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { ops: [validOp, invalidOp], clientId }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json().results).toEqual([ + expect.objectContaining({ opId: validOp.id, accepted: true }), + expect.objectContaining({ + opId: invalidOp.id, + accepted: false, + errorCode: SYNC_ERROR_CODES.INVALID_SCHEMA_VERSION, + }), + ]); + expect(mocks.syncService.uploadOps).toHaveBeenCalledWith(1, clientId, [ + validOp, + invalidOp, + ]); + }); + it('should subtract exact already-stored duplicate ops from the ops quota gate', async () => { const clientId = 'known-duplicate-quota-client'; const duplicateOp = { diff --git a/packages/sync-core/src/remote-apply.ts b/packages/sync-core/src/remote-apply.ts index 9676c3a311..70544f8d81 100644 --- a/packages/sync-core/src/remote-apply.ts +++ b/packages/sync-core/src/remote-apply.ts @@ -113,20 +113,58 @@ export const applyRemoteOperations = async < } }); + let reducerCommitCallbackCount = 0; + let reducerCommitPromise: Promise | 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}.`, diff --git a/packages/sync-core/tests/remote-apply.spec.ts b/packages/sync-core/tests/remote-apply.spec.ts index 3c71023caa..bdfedb28db 100644 --- a/packages/sync-core/tests/remote-apply.spec.ts +++ b/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> = { + 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> = { + 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> = { + 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 => diff --git a/src/app/op-log/sync/immediate-upload.service.spec.ts b/src/app/op-log/sync/immediate-upload.service.spec.ts index 6d18e1ce4a..15a4880743 100644 --- a/src/app/op-log/sync/immediate-upload.service.spec.ts +++ b/src/app/op-log/sync/immediate-upload.service.spec.ts @@ -205,6 +205,49 @@ describe('ImmediateUploadService', () => { // Piggybacked ops are processed internally, no checkmark shown expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalled(); })); + + it('should report ERROR when the local-win follow-up is blocked', fakeAsync(() => { + mockSyncService.uploadPendingOps.and.returnValues( + Promise.resolve(completedResult({ uploadedCount: 1, localWinOpsCreated: 1 })), + Promise.resolve({ kind: 'blocked_incompatible' }), + ); + + service.initialize(); + service.trigger(); + tick(2000); + flush(); + + expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR'); + expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalledWith('IN_SYNC'); + })); + + it('should not show a checkmark when the local-win follow-up is cancelled', fakeAsync(() => { + mockSyncService.uploadPendingOps.and.returnValues( + Promise.resolve(completedResult({ uploadedCount: 1, localWinOpsCreated: 1 })), + Promise.resolve({ kind: 'cancelled' }), + ); + + service.initialize(); + service.trigger(); + tick(2000); + flush(); + + expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalled(); + })); + + it('should defer the checkmark when the local-win follow-up receives piggybacked ops', fakeAsync(() => { + mockSyncService.uploadPendingOps.and.returnValues( + Promise.resolve(completedResult({ uploadedCount: 1, localWinOpsCreated: 1 })), + Promise.resolve(completedResult({ uploadedCount: 1, piggybackedOpsCount: 1 })), + ); + + service.initialize(); + service.trigger(); + tick(2000); + flush(); + + expect(mockProviderManager.setSyncStatus).not.toHaveBeenCalled(); + })); }); // #8309: the immediate-upload side channel must not interleave with another diff --git a/src/app/op-log/sync/immediate-upload.service.ts b/src/app/op-log/sync/immediate-upload.service.ts index 22c0fc0c9b..5f1c4365ca 100644 --- a/src/app/op-log/sync/immediate-upload.service.ts +++ b/src/app/op-log/sync/immediate-upload.service.ts @@ -249,11 +249,31 @@ export class ImmediateUploadService implements OnDestroy { // If LWW local-wins created new update ops from piggybacked ops, // do a follow-up upload to push them to the server immediately + let finalResult = result; + let totalUploadedCount = result.uploadedCount; + let hasPermanentRejection = result.permanentRejectionCount > 0; if (result.localWinOpsCreated > 0) { OpLog.verbose( `ImmediateUploadService: LWW created ${result.localWinOpsCreated} local-win op(s), re-uploading`, ); - await this._syncService.uploadPendingOps(syncCapableProvider); + const followUpResult = + await this._syncService.uploadPendingOps(syncCapableProvider); + if (followUpResult.kind === 'blocked_incompatible') { + OpLog.warn( + 'ImmediateUploadService: Local-win follow-up blocked by an incompatible operation', + ); + this._providerManager.setSyncStatus('ERROR'); + return; + } + if ( + followUpResult.kind === 'cancelled' || + followUpResult.kind === 'blocked_fresh_client' + ) { + return; + } + finalResult = followUpResult; + totalUploadedCount += followUpResult.uploadedCount; + hasPermanentRejection ||= followUpResult.permanentRejectionCount > 0; } // Read the validation latch BEFORE any IN_SYNC / deferred-checkmark @@ -270,20 +290,29 @@ export class ImmediateUploadService implements OnDestroy { // Don't show checkmark when piggybacked ops exist - there may be more // remote ops pending. Let normal sync cycle confirm full sync state. - if (result.piggybackedOpsCount > 0) { + if (hasPermanentRejection) { + this._providerManager.setSyncStatus('ERROR'); + return; + } + + if ( + finalResult.piggybackedOpsCount > 0 || + finalResult.hasMorePiggyback || + finalResult.localWinOpsCreated > 0 + ) { OpLog.verbose( - `ImmediateUploadService: Uploaded ${result.uploadedCount} ops, ` + - `processed ${result.piggybackedOpsCount} piggybacked (checkmark deferred)`, + `ImmediateUploadService: Uploaded ${totalUploadedCount} ops, ` + + `processed ${finalResult.piggybackedOpsCount} piggybacked (checkmark deferred)`, ); return; } // Show checkmark ONLY when server confirms no pending remote ops // (empty piggybackedOps means we're confirmed in sync) - if (result.uploadedCount > 0 || result.localWinOpsCreated > 0) { + if (totalUploadedCount > 0) { this._providerManager.setSyncStatus('IN_SYNC'); OpLog.verbose( - `ImmediateUploadService: Uploaded ${result.uploadedCount} ops, confirmed in sync`, + `ImmediateUploadService: Uploaded ${totalUploadedCount} ops, confirmed in sync`, ); } } catch (e) {