diff --git a/docs/long-term-plans/sync-core-extraction-plan.md b/docs/long-term-plans/sync-core-extraction-plan.md index 0c012fb397..3de9634059 100644 --- a/docs/long-term-plans/sync-core-extraction-plan.md +++ b/docs/long-term-plans/sync-core-extraction-plan.md @@ -5,8 +5,9 @@ > pure helper slices are present on this branch. PR 4a port contracts have > started with the app-side replay/operation-store services explicitly satisfying > sync-core interfaces and adapter contract specs covering the initial ports. -> Remaining cleanup is PR text alignment plus targeted future `SyncLogger` -> routing for files as they move.** +> PR 4b has started with the remote-apply crash-safety coordinator moved behind +> ports. Remaining cleanup is PR text alignment plus targeted future +> `SyncLogger` routing for files as they move.** **Goal:** Carve the sync engine out of `src/app/op-log/` into a reusable, framework-agnostic, **domain-agnostic** `@sp/sync-core` package, plus a sibling @@ -729,6 +730,23 @@ service remains app-side. Move only orchestration code whose dependencies are already represented by ports and whose behavior can be tested without Angular. +Status: started on this branch. `@sp/sync-core` now exports +`applyRemoteOperations()` plus the narrow `RemoteOperationApplyStorePort`. +`RemoteOpsProcessingService.applyNonConflictingOps()` delegates the generic +remote-apply crash-safety ordering to that coordinator: + +1. append incoming remote ops as pending while atomically skipping duplicates; +2. apply only newly appended ops through `OperationApplyPort`; +3. mark applied seqs; +4. merge applied remote vector clocks; +5. clear older full-state ops after a newer applied full-state op lands; +6. mark the failed op and remaining unapplied ops as failed on partial apply + errors. + +The Angular service still owns app diagnostics, validation/session latching, +snack notifications, conflict detection, NgRx dispatch construction, and the +IndexedDB implementation. + ### Candidate Moves - Upload batching/retry logic from `OperationLogUploadService` if provider and diff --git a/packages/sync-core/src/index.ts b/packages/sync-core/src/index.ts index 8ba02a442a..d3e6b38a35 100644 --- a/packages/sync-core/src/index.ts +++ b/packages/sync-core/src/index.ts @@ -74,6 +74,15 @@ export type { LwwUpdateActionTypeHelpers } from './lww-update-action-types'; // Apply-operation result and option types. export type { ApplyOperationsResult, ApplyOperationsOptions } from './apply.types'; +// Remote operation application coordinator. +export { applyRemoteOperations } from './remote-apply'; +export type { + ApplyRemoteOperationsOptions, + RemoteApplyOperationsResult, + RemoteOperationAppendResult, + RemoteOperationApplyStorePort, +} from './remote-apply'; + // Port contracts for app-side orchestration adapters. export type { ActionDispatchPort, diff --git a/packages/sync-core/src/remote-apply.ts b/packages/sync-core/src/remote-apply.ts new file mode 100644 index 0000000000..a47fe72c97 --- /dev/null +++ b/packages/sync-core/src/remote-apply.ts @@ -0,0 +1,159 @@ +import type { ApplyOperationsResult } from './apply.types'; +import type { Operation } from './operation.types'; +import type { OperationApplyPort } from './ports'; + +export interface RemoteOperationAppendResult< + TOperation extends Operation = Operation, +> { + seqs: number[]; + writtenOps: TOperation[]; + skippedCount: number; +} + +/** + * Store methods needed by the generic remote-apply coordinator. + * + * This stays narrower than a full operation-log database service and only + * represents the crash-safety transitions around applying remote operations. + */ +export interface RemoteOperationApplyStorePort< + TOperation extends Operation = Operation, +> { + appendBatchSkipDuplicates( + ops: TOperation[], + source: 'remote', + options: { pendingApply: true }, + ): Promise>; + markApplied(seqs: number[]): Promise; + markFailed(opIds: string[]): Promise; + mergeRemoteOpClocks(ops: TOperation[]): Promise; + clearFullStateOpsExcept(excludeIds: string[]): Promise; +} + +export interface ApplyRemoteOperationsOptions< + TOperation extends Operation = Operation, +> { + ops: TOperation[]; + store: RemoteOperationApplyStorePort; + applier: OperationApplyPort; + isFullStateOperation?: (op: TOperation) => boolean; +} + +export interface RemoteApplyOperationsResult< + TOperation extends Operation = Operation, +> { + appendedOps: TOperation[]; + skippedCount: number; + appliedOps: TOperation[]; + appliedSeqs: number[]; + clearedFullStateOpCount: number; + failedOp?: ApplyOperationsResult['failedOp']; + failedOpIds: string[]; +} + +const emptyRemoteApplyResult = < + TOperation extends Operation, +>(): RemoteApplyOperationsResult => ({ + appendedOps: [], + skippedCount: 0, + appliedOps: [], + appliedSeqs: [], + clearedFullStateOpCount: 0, + failedOpIds: [], +}); + +/** + * Applies remote operations through host ports and records crash-safety state. + * + * Host apps keep framework-specific dispatch, storage, diagnostics, validation, + * and user notifications outside the package. This coordinator only enforces + * 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; + * 5. retain only the newest applied full-state ops when configured; + * 6. mark the failed op and remaining unapplied ops as failed on partial error. + */ +export const applyRemoteOperations = async < + TOperation extends Operation = Operation, +>({ + ops, + store, + applier, + isFullStateOperation = () => false, +}: ApplyRemoteOperationsOptions): Promise< + RemoteApplyOperationsResult +> => { + if (ops.length === 0) { + return emptyRemoteApplyResult(); + } + + const appendResult = await store.appendBatchSkipDuplicates(ops, 'remote', { + pendingApply: true, + }); + + if (appendResult.writtenOps.length === 0) { + return { + ...emptyRemoteApplyResult(), + skippedCount: appendResult.skippedCount, + }; + } + + const opIdToSeq = new Map(); + appendResult.writtenOps.forEach((op, index) => { + const seq = appendResult.seqs[index]; + if (seq !== undefined) { + opIdToSeq.set(op.id, seq); + } + }); + + const applyResult = await applier.applyOperations(appendResult.writtenOps); + const appliedSeqs = applyResult.appliedOps + .map((op) => opIdToSeq.get(op.id)) + .filter((seq): seq is number => seq !== undefined); + + let clearedFullStateOpCount = 0; + + if (appliedSeqs.length > 0) { + await store.markApplied(appliedSeqs); + await store.mergeRemoteOpClocks(applyResult.appliedOps); + + const appliedFullStateOpIds = applyResult.appliedOps + .filter(isFullStateOperation) + .map((op) => op.id); + + if (appliedFullStateOpIds.length > 0) { + clearedFullStateOpCount = + await store.clearFullStateOpsExcept(appliedFullStateOpIds); + } + } + + const failedOpIds = applyResult.failedOp + ? getFailedOpIds(appendResult.writtenOps, applyResult.failedOp.op) + : []; + + if (failedOpIds.length > 0) { + await store.markFailed(failedOpIds); + } + + return { + appendedOps: appendResult.writtenOps, + skippedCount: appendResult.skippedCount, + appliedOps: applyResult.appliedOps, + appliedSeqs, + clearedFullStateOpCount, + failedOp: applyResult.failedOp, + failedOpIds, + }; +}; + +const getFailedOpIds = >( + opsToApply: TOperation[], + failedOp: TOperation, +): string[] => { + const failedOpIndex = opsToApply.findIndex((op) => op.id === failedOp.id); + const failedOps = failedOpIndex === -1 ? [failedOp] : opsToApply.slice(failedOpIndex); + return failedOps.map((op) => op.id); +}; diff --git a/packages/sync-core/tests/remote-apply.spec.ts b/packages/sync-core/tests/remote-apply.spec.ts new file mode 100644 index 0000000000..b2aca4f786 --- /dev/null +++ b/packages/sync-core/tests/remote-apply.spec.ts @@ -0,0 +1,149 @@ +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'; + +const createOperation = (id: string, opType = 'UPD'): Operation => ({ + id, + actionType: '[Test] Action', + opType, + entityType: 'TASK', + entityId: id, + payload: {}, + clientId: 'client-1', + vectorClock: { client1: 1 }, + timestamp: 1, + schemaVersion: 1, +}); + +const createStore = (appendResult: { + seqs: number[]; + writtenOps: Operation[]; + skippedCount: number; +}): RemoteOperationApplyStorePort> => ({ + appendBatchSkipDuplicates: vi.fn().mockResolvedValue(appendResult), + markApplied: vi.fn().mockResolvedValue(undefined), + markFailed: vi.fn().mockResolvedValue(undefined), + mergeRemoteOpClocks: vi.fn().mockResolvedValue(undefined), + clearFullStateOpsExcept: vi.fn().mockResolvedValue(0), +}); + +const createApplier = ( + result: Awaited>['applyOperations']>>, +): OperationApplyPort> => ({ + applyOperations: vi.fn().mockResolvedValue(result), +}); + +describe('applyRemoteOperations', () => { + it('appends remote ops as pending and applies only written ops', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const store = createStore({ seqs: [10], writtenOps: [op2], skippedCount: 1 }); + const applier = createApplier({ appliedOps: [op2] }); + + const result = await applyRemoteOperations({ + ops: [op1, op2], + store, + applier, + }); + + expect(store.appendBatchSkipDuplicates).toHaveBeenCalledWith([op1, op2], 'remote', { + pendingApply: true, + }); + expect(applier.applyOperations).toHaveBeenCalledWith([op2]); + expect(store.markApplied).toHaveBeenCalledWith([10]); + expect(store.mergeRemoteOpClocks).toHaveBeenCalledWith([op2]); + expect(result).toEqual({ + appendedOps: [op2], + skippedCount: 1, + appliedOps: [op2], + appliedSeqs: [10], + clearedFullStateOpCount: 0, + failedOpIds: [], + }); + }); + + it('does not apply when every incoming op is skipped as a duplicate', async () => { + const op = createOperation('op-1'); + const store = createStore({ seqs: [], writtenOps: [], skippedCount: 1 }); + const applier = createApplier({ appliedOps: [] }); + + const result = await applyRemoteOperations({ ops: [op], store, applier }); + + expect(applier.applyOperations).not.toHaveBeenCalled(); + expect(store.markApplied).not.toHaveBeenCalled(); + expect(store.mergeRemoteOpClocks).not.toHaveBeenCalled(); + expect(result).toEqual({ + appendedOps: [], + skippedCount: 1, + appliedOps: [], + appliedSeqs: [], + clearedFullStateOpCount: 0, + failedOpIds: [], + }); + }); + + it('clears older full-state ops after successfully applying a full-state op', async () => { + const fullStateOp = createOperation('sync-import-1', 'SYNC_IMPORT'); + const store = createStore({ + seqs: [1], + writtenOps: [fullStateOp], + skippedCount: 0, + }); + vi.mocked(store.clearFullStateOpsExcept).mockResolvedValue(3); + const applier = createApplier({ appliedOps: [fullStateOp] }); + + const result = await applyRemoteOperations({ + ops: [fullStateOp], + store, + applier, + isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT', + }); + + expect(store.clearFullStateOpsExcept).toHaveBeenCalledWith(['sync-import-1']); + expect(result.clearedFullStateOpCount).toBe(3); + }); + + it('marks the failed op and remaining unapplied ops as failed', async () => { + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const op3 = createOperation('op-3'); + const error = new Error('archive failure'); + const store = createStore({ + seqs: [1, 2, 3], + writtenOps: [op1, op2, op3], + skippedCount: 0, + }); + const applier = createApplier({ + appliedOps: [op1], + failedOp: { op: op2, error }, + }); + + const result = await applyRemoteOperations({ + ops: [op1, op2, op3], + store, + applier, + }); + + expect(store.markApplied).toHaveBeenCalledWith([1]); + expect(store.mergeRemoteOpClocks).toHaveBeenCalledWith([op1]); + expect(store.markFailed).toHaveBeenCalledWith(['op-2', 'op-3']); + expect(result.failedOp).toEqual({ op: op2, error }); + expect(result.failedOpIds).toEqual(['op-2', 'op-3']); + }); + + it('marks only the reported failed op if it is not in the appended batch', async () => { + const op1 = createOperation('op-1'); + const unknownFailedOp = createOperation('unknown-failed-op'); + const store = createStore({ seqs: [1], writtenOps: [op1], skippedCount: 0 }); + const applier = createApplier({ + appliedOps: [], + failedOp: { op: unknownFailedOp, error: new Error('unexpected') }, + }); + + const result = await applyRemoteOperations({ ops: [op1], store, applier }); + + expect(store.markFailed).toHaveBeenCalledWith(['unknown-failed-op']); + expect(result.failedOpIds).toEqual(['unknown-failed-op']); + }); +}); diff --git a/src/app/op-log/persistence/operation-log-store.service.ts b/src/app/op-log/persistence/operation-log-store.service.ts index e91bf89af4..5028e1e335 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -1,5 +1,5 @@ import { inject, Injectable } from '@angular/core'; -import type { OperationStorePort } from '@sp/sync-core'; +import type { OperationStorePort, RemoteOperationApplyStorePort } from '@sp/sync-core'; import { DBSchema, IDBPDatabase, openDB } from 'idb'; import { Operation, @@ -171,10 +171,11 @@ interface OpLogDB extends DBSchema { @Injectable({ providedIn: 'root', }) -export class OperationLogStoreService implements OperationStorePort< - Operation, - OperationLogEntry -> { +export class OperationLogStoreService + implements + OperationStorePort, + RemoteOperationApplyStorePort +{ private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); private _db?: IDBPDatabase; private _initPromise?: Promise; diff --git a/src/app/op-log/sync/remote-ops-processing.service.ts b/src/app/op-log/sync/remote-ops-processing.service.ts index 21488b2df6..f0e467b267 100644 --- a/src/app/op-log/sync/remote-ops-processing.service.ts +++ b/src/app/op-log/sync/remote-ops-processing.service.ts @@ -1,4 +1,5 @@ import { inject, Injectable } from '@angular/core'; +import { applyRemoteOperations } from '@sp/sync-core'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { ConflictResult, @@ -385,166 +386,15 @@ export class RemoteOpsProcessingService { ops: Operation[], callerHoldsLock: boolean = false, ): Promise { - // Map op ID to seq for marking partial success - const opIdToSeq = new Map(); + await this._logFullStateApplyDiagnostics(ops); - // Atomically filter and append ops in a single transaction (issue #6343). - // This eliminates the TOCTOU race between filterNewOps() and appendBatch() - // that caused persistent "Duplicate operation detected" errors. - const opsToApply = await this._filterAndAppendOps(ops, opIdToSeq); - - // Apply only NON-duplicate ops to NgRx store - if (opsToApply.length > 0) { - const result = await this.operationApplier.applyOperations(opsToApply); - - // Mark successfully applied ops - const appliedSeqs = result.appliedOps - .map((op) => opIdToSeq.get(op.id)) - .filter((seq): seq is number => seq !== undefined); - - if (appliedSeqs.length > 0) { - await this.opLogStore.markApplied(appliedSeqs); - - // CRITICAL: Merge remote ops' vector clocks into local clock. - // This ensures subsequent local operations have clocks that "dominate" - // the remote ops (GREATER_THAN instead of CONCURRENT). - // Without this, ops created after a SYNC_IMPORT would be incorrectly - // filtered by SyncImportFilterService as "invalidated by import". - await this.opLogStore.mergeRemoteOpClocks(result.appliedOps); - - // Check if a full-state op was applied, and clear older full-state ops - const appliedFullStateOp = result.appliedOps.find( - (op) => - op.opType === OpType.SyncImport || - op.opType === OpType.BackupImport || - op.opType === OpType.Repair, - ); - if (appliedFullStateOp) { - // CRITICAL FIX: Clear older full-state ops AFTER successfully storing the new one. - // This prevents the scenario where: - // 1. Client A has old SYNC_IMPORT from client X with minimal clock {X:1} - // 2. Client B uploads new SYNC_IMPORT with its own minimal clock - // 3. Client A downloads and stores B's SYNC_IMPORT - // 4. Without clearing, getLatestFullStateOpEntry might return X's old import - // (if it has a higher UUIDv7 timestamp) - // 5. New operations appear CONCURRENT with X's import and get filtered - // - // We clear AFTER storing (not before) to ensure crash safety. - // We exclude the newly stored full-state op IDs so we don't delete what we just added. - const newFullStateOpIds = result.appliedOps - .filter( - (op) => - op.opType === OpType.SyncImport || - op.opType === OpType.BackupImport || - op.opType === OpType.Repair, - ) - .map((op) => op.id); - const clearedCount = - await this.opLogStore.clearFullStateOpsExcept(newFullStateOpIds); - if (clearedCount > 0) { - OpLog.normal( - `RemoteOpsProcessingService: Cleared ${clearedCount} old full-state op(s) after applying new one.`, - ); - } - } - - OpLog.normal( - `RemoteOpsProcessingService: Applied and marked ${appliedSeqs.length} remote ops`, - ); - } - - // Handle partial failure - if (result.failedOp) { - // Find all ops that weren't applied (failed op + remaining ops) - const failedOpIndex = opsToApply.findIndex( - (op) => op.id === result.failedOp!.op.id, - ); - const failedOps = opsToApply.slice(failedOpIndex); - const failedOpIds = failedOps.map((op) => op.id); - - OpLog.err( - `RemoteOpsProcessingService: ${result.appliedOps.length} ops applied before failure. ` + - `Marking ${failedOpIds.length} ops as failed.`, - result.failedOp.error, - ); - await this.opLogStore.markFailed(failedOpIds); - - await this._validateAndFlagSession( - 'partial-apply-failure', - callerHoldsLock, - 'RemoteOpsProcessingService: State validation failed after partial apply failure', - ); - - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.PARTIAL_APPLY_FAILURE, - }); - - // Re-throw if it's a SyncStateCorruptedError, otherwise wrap it - throw result.failedOp.error; - } - } - } - - /** - * Atomically filters out already-applied ops and appends new ones to the store. - * Uses appendBatchSkipDuplicates() to check and insert within a single IndexedDB - * transaction, eliminating the TOCTOU race condition (issue #6343). - * - * @param ops - Operations to filter and potentially append - * @param opIdToSeq - Map to populate with op ID -> sequence number mappings - * @returns The operations that were actually appended (after filtering) - */ - private async _filterAndAppendOps( - ops: Operation[], - opIdToSeq: Map, - ): Promise { - // DIAGNOSTIC: Check if any full-state ops will be applied - const fullStateOps = ops.filter( - (op) => - op.opType === OpType.SyncImport || - op.opType === OpType.BackupImport || - op.opType === OpType.Repair, - ); - if (fullStateOps.length > 0) { - // Snapshot the receiver's prior clock and unsynced-op tally before the - // batch lands. After append, the receiver's state advances and we can no - // longer reconstruct what was about to be wiped. Captures both the prior - // vector clock (whose entries the SYNC_IMPORT will collapse) and the - // count of local unsynced user work that the SyncImportFilter will then - // discard — that count answers the "post-import edits failed to upload - // vs. dropped by the filter" question on the next incident. - const priorClock = await this.opLogStore.getVectorClock(); - const priorUnsynced = await this.opLogStore.getUnsynced(); - const priorUnsyncedByOpType = priorUnsynced.reduce>( - (acc, entry) => { - const key = entry.op.opType; - acc[key] = (acc[key] ?? 0) + 1; - return acc; - }, - {}, - ); - OpLog.log( - `RemoteOpsProcessingService: APPLYING FULL-STATE OP(s): ${fullStateOps.map((op) => `${op.opType} from ${op.clientId}`).join(', ')}`, - { - incoming: fullStateOps.map((op) => ({ - opType: op.opType, - clientId: op.clientId, - syncImportReason: op.syncImportReason ?? null, - vectorClock: op.vectorClock, - })), - priorClock: priorClock ?? null, - priorClockSize: priorClock ? Object.keys(priorClock).length : 0, - priorUnsyncedCount: priorUnsynced.length, - priorUnsyncedByOpType, - }, - ); - } - - // Atomically check-and-insert within a single transaction (issue #6343). - // Duplicates are silently skipped rather than causing ConstraintError. - const result = await this.opLogStore.appendBatchSkipDuplicates(ops, 'remote', { - pendingApply: true, + // Core owns the generic crash-safety ordering. Angular diagnostics, + // validation, and user notifications stay in this service. + const result = await applyRemoteOperations({ + ops, + store: this.opLogStore, + applier: this.operationApplier, + isFullStateOperation: this._isFullStateOperation, }); if (result.skippedCount > 0) { @@ -553,8 +403,92 @@ export class RemoteOpsProcessingService { ); } - result.writtenOps.forEach((op, i) => opIdToSeq.set(op.id, result.seqs[i])); - return result.writtenOps; + if (result.clearedFullStateOpCount > 0) { + OpLog.normal( + `RemoteOpsProcessingService: Cleared ${result.clearedFullStateOpCount} old full-state op(s) after applying new one.`, + ); + } + + if (result.appliedSeqs.length > 0) { + OpLog.normal( + `RemoteOpsProcessingService: Applied and marked ${result.appliedSeqs.length} remote ops`, + ); + } + + // Handle partial failure + if (result.failedOp) { + OpLog.err( + `RemoteOpsProcessingService: ${result.appliedOps.length} ops applied before failure. ` + + `Marking ${result.failedOpIds.length} ops as failed.`, + result.failedOp.error, + ); + + await this._validateAndFlagSession( + 'partial-apply-failure', + callerHoldsLock, + 'RemoteOpsProcessingService: State validation failed after partial apply failure', + ); + + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.PARTIAL_APPLY_FAILURE, + }); + + // Re-throw if it's a SyncStateCorruptedError, otherwise wrap it + throw result.failedOp.error; + } + } + + private _isFullStateOperation(op: Operation): boolean { + return ( + op.opType === OpType.SyncImport || + op.opType === OpType.BackupImport || + op.opType === OpType.Repair + ); + } + + /** + * Logs receiver state before full-state remote ops land. This diagnostic is + * app-side because it reads SP's operation store and uses exportable app logs. + */ + private async _logFullStateApplyDiagnostics(ops: Operation[]): Promise { + const fullStateOps = ops.filter(this._isFullStateOperation); + if (fullStateOps.length === 0) { + return; + } + + // Snapshot the receiver's prior clock and unsynced-op tally before the + // batch lands. After append, the receiver's state advances and we can no + // longer reconstruct what was about to be wiped. Captures both the prior + // vector clock (whose entries the SYNC_IMPORT will collapse) and the + // count of local unsynced user work that the SyncImportFilter will then + // discard — that count answers the "post-import edits failed to upload + // vs. dropped by the filter" question on the next incident. + const priorClock = await this.opLogStore.getVectorClock(); + const priorUnsynced = await this.opLogStore.getUnsynced(); + const priorUnsyncedByOpType = priorUnsynced.reduce>( + (acc, entry) => { + const key = entry.op.opType; + acc[key] = (acc[key] ?? 0) + 1; + return acc; + }, + {}, + ); + OpLog.log( + `RemoteOpsProcessingService: APPLYING FULL-STATE OP(s): ${fullStateOps.map((op) => `${op.opType} from ${op.clientId}`).join(', ')}`, + { + incoming: fullStateOps.map((op) => ({ + opType: op.opType, + clientId: op.clientId, + syncImportReason: op.syncImportReason ?? null, + vectorClock: op.vectorClock, + })), + priorClock: priorClock ?? null, + priorClockSize: priorClock ? Object.keys(priorClock).length : 0, + priorUnsyncedCount: priorUnsynced.length, + priorUnsyncedByOpType, + }, + ); } /**