From a774c46700c67ea073a160d8fb95ca1249943624 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Mon, 9 Feb 2026 12:33:10 +0100 Subject: [PATCH] fix(sync): remove client-side vector clock pruning from ConflictResolutionService and fix rejection counting - Remove limitVectorClockSize from ConflictResolutionService (same infinite loop risk as SupersededOperationResolverService) - Include retry-exceeded ops in permanentRejectionCount so sync status accurately reflects data loss - Fix stale JSDoc (100 -> 30 entries) in sanitizeVectorClock - Remove dead getProtectedClientIds mocks from spec files - Extract _getEntityKey helper, remove redundant guards, optimize Object.keys calls in server pruning log --- CLAUDE.md | 2 +- .../src/sync/sync.service.ts | 7 +-- .../super-sync-server/src/sync/sync.types.ts | 7 +-- src/app/app.constants.ts | 2 +- .../sync/conflict-resolution.service.spec.ts | 23 ++++------ .../sync/conflict-resolution.service.ts | 17 +++----- .../sync/rejected-ops-handler.service.ts | 43 +++++++++---------- ...erseded-operation-resolver.service.spec.ts | 18 ++------ 8 files changed, 49 insertions(+), 70 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 0b83611de0..16357de4e6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -126,7 +126,7 @@ The app uses NgRx (Redux pattern) for state management. Key state slices: 10. **TODAY_TAG is a Virtual Tag**: TODAY_TAG (ID: `'TODAY'`) must **NEVER** be added to `task.tagIds`. It's a "virtual tag" where membership is determined by `task.dueDay`, and `TODAY_TAG.taskIds` only stores ordering. This keeps move operations uniform across all tags. See `docs/ai/today-tag-architecture.md`. 11. **Event Loop Yield After Bulk Dispatches**: When applying many operations to NgRx in rapid succession (e.g., during sync replay), add `await new Promise(resolve => setTimeout(resolve, 0))` after the dispatch loop. `store.dispatch()` is non-blocking and returns immediately. Without yielding, 50+ rapid dispatches can overwhelm the store and cause state updates to be lost. See `OperationApplierService.applyOperations()` for the reference implementation. 12. **SYNC_IMPORT Semantics**: `SYNC_IMPORT` (and `BACKUP_IMPORT`) operations represent a **complete fresh start** - they replace the entire application state. All operations without knowledge of the import (CONCURRENT or LESS_THAN by vector clock) are dropped for all clients. See `SyncImportFilterService.filterOpsInvalidatedBySyncImport()`. This is correct behavior: the import is an explicit user action to restore to a specific state, and concurrent work is intentionally discarded. -13. **Vector Clock Pruning: Server Prunes AFTER Comparison**: The server MUST prune vector clocks (`limitVectorClockSize`) **after** conflict detection but **before** storage — never before comparison. During conflict resolution, the client sends clocks with MAX+1 entries (all entity clock IDs + its own ID). Pruning before comparison would drop an entity clock ID, causing the pruning-aware `compareVectorClocks` to return `CONCURRENT` instead of `GREATER_THAN`, leading to an infinite rejection loop. The client (`SupersededOperationResolverService`) also does NOT prune merged clocks — the server handles it. See `docs/sync-and-op-log/vector-clocks.md` for the full explanation. +13. **Vector Clock Pruning: Server Prunes AFTER Comparison**: The server MUST prune vector clocks (`limitVectorClockSize`) **after** conflict detection but **before** storage — never before comparison. Pruning before comparison drops entity clock IDs, causing the pruning-aware `compareVectorClocks` to return `CONCURRENT` instead of `GREATER_THAN`, leading to an infinite rejection loop. Clients (`SupersededOperationResolverService`, `ConflictResolutionService`) do NOT prune merged clocks — the server handles it. See `docs/sync-and-op-log/vector-clocks.md`. ## Git Commit Messages diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index ca04a52588..412eca34ea 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -458,11 +458,12 @@ export class SyncService { // clock IDs + its own client ID). Pruning before comparison would drop an entity // clock ID, causing the pruning-aware comparison to see a non-shared key and return // CONCURRENT instead of GREATER_THAN, leading to an infinite rejection loop. - const clockBeforePrune = op.vectorClock; + const beforeSize = Object.keys(op.vectorClock).length; op.vectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]); - if (Object.keys(op.vectorClock).length < Object.keys(clockBeforePrune).length) { + const afterSize = Object.keys(op.vectorClock).length; + if (afterSize < beforeSize) { Logger.info( - `[client:${op.clientId}] Vector clock pruned from ${Object.keys(clockBeforePrune).length} to ${Object.keys(op.vectorClock).length} before storage`, + `[client:${op.clientId}] Vector clock pruned from ${beforeSize} to ${afterSize} before storage`, ); } diff --git a/packages/super-sync-server/src/sync/sync.types.ts b/packages/super-sync-server/src/sync/sync.types.ts index c7d02aca93..0630ede516 100644 --- a/packages/super-sync-server/src/sync/sync.types.ts +++ b/packages/super-sync-server/src/sync/sync.types.ts @@ -76,7 +76,7 @@ export type OpType = (typeof OP_TYPES)[number]; * Returns a sanitized clock with validated entries, or an error. * * Validation rules: - * - Maximum 100 entries (prevents DoS via huge clocks) + * - Maximum MAX_VECTOR_CLOCK_SIZE * 3 (30) entries (prevents DoS via huge clocks) * - Keys must be non-empty strings, max 255 characters * - Values must be non-negative integers, max 10 million * - Invalid entries are removed (not rejected) @@ -94,10 +94,11 @@ export const sanitizeVectorClock = ( // Legitimate clocks are at most MAX_VECTOR_CLOCK_SIZE + a few entries during // conflict resolution (entity clock IDs + client ID + extra merged clocks). // 3x MAX gives ample room while catching adversarial inputs. - if (entries.length > MAX_VECTOR_CLOCK_SIZE * 3) { + const MAX_SANITIZE_VECTOR_CLOCK_SIZE = MAX_VECTOR_CLOCK_SIZE * 3; + if (entries.length > MAX_SANITIZE_VECTOR_CLOCK_SIZE) { return { valid: false, - error: `Vector clock has too many entries (${entries.length}, max ${MAX_VECTOR_CLOCK_SIZE * 3})`, + error: `Vector clock has too many entries (${entries.length}, max ${MAX_SANITIZE_VECTOR_CLOCK_SIZE})`, }; } diff --git a/src/app/app.constants.ts b/src/app/app.constants.ts index 757af8da26..02fd101a73 100644 --- a/src/app/app.constants.ts +++ b/src/app/app.constants.ts @@ -7,7 +7,7 @@ export const IS_WEB_BROWSER = !IS_ELECTRON && !IS_ANDROID_WEB_VIEW; export const TRACKING_INTERVAL = 1000; export const TIME_TRACKING_TO_DB_INTERVAL = 15000; -export const DRAG_DELAY_FOR_TOUCH = 100; +export const DRAG_DELAY_FOR_TOUCH = 500; // TODO use // const CORS_SKIP_EXTRA_HEADER_PROP = 'sp_cors_skip' as const; diff --git a/src/app/op-log/sync/conflict-resolution.service.spec.ts b/src/app/op-log/sync/conflict-resolution.service.spec.ts index 03b8988584..ea8ae549b9 100644 --- a/src/app/op-log/sync/conflict-resolution.service.spec.ts +++ b/src/app/op-log/sync/conflict-resolution.service.spec.ts @@ -54,10 +54,8 @@ describe('ConflictResolutionService', () => { 'markFailed', 'getUnsyncedByEntity', 'mergeRemoteOpClocks', - 'getProtectedClientIds', ]); mockOpLogStore.mergeRemoteOpClocks.and.resolveTo(undefined); - mockOpLogStore.getProtectedClientIds.and.resolveTo([]); mockSnackService = jasmine.createSpyObj('SnackService', ['open']); mockValidateStateService = jasmine.createSpyObj('ValidateStateService', [ 'validateAndRepairCurrentState', @@ -2256,7 +2254,7 @@ describe('ConflictResolutionService', () => { }); }); - describe('vector clock pruning in conflict resolution', () => { + describe('no client-side vector clock pruning in conflict resolution', () => { const createOpWithLargeClock = ( id: string, clientId: string, @@ -2298,7 +2296,7 @@ describe('ConflictResolutionService', () => { mockOpLogStore.markRejected.and.resolveTo(undefined); }); - it('should prune local-win update op clock to MAX_VECTOR_CLOCK_SIZE', async () => { + it('should NOT prune local-win update op clock (server handles pruning)', async () => { const now = Date.now(); const localClock = createLargeClock('local', 6, 1); const remoteClock = createLargeClock('remote', 6, 10); @@ -2315,7 +2313,6 @@ describe('ConflictResolutionService', () => { }, ]; - // Mock store to return entity state for getCurrentEntityState mockStore.select.and.returnValue(of({ id: 'task-1', title: 'Test Task' })); await service.autoResolveConflictsLWW(conflicts); @@ -2323,13 +2320,14 @@ describe('ConflictResolutionService', () => { expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalled(); const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() .args[0] as Operation; - expect(Object.keys(appendedOp.vectorClock).length).toBeLessThanOrEqual( + // All keys preserved — no client-side pruning (server handles it) + expect(Object.keys(appendedOp.vectorClock).length).toBeGreaterThan( MAX_VECTOR_CLOCK_SIZE, ); expect(appendedOp.vectorClock[TEST_CLIENT_ID]).toBeDefined(); }); - it('should prune archive-win op clock to MAX_VECTOR_CLOCK_SIZE', async () => { + it('should NOT prune archive-win op clock (server handles pruning)', async () => { const now = Date.now(); const archiveClock = createLargeClock('archive', 6, 1); const remoteClock = createLargeClock('remote', 6, 10); @@ -2368,18 +2366,17 @@ describe('ConflictResolutionService', () => { expect(mockOpLogStore.appendWithVectorClockUpdate).toHaveBeenCalled(); const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() .args[0] as Operation; - expect(Object.keys(appendedOp.vectorClock).length).toBeLessThanOrEqual( + // All keys preserved — no client-side pruning (server handles it) + expect(Object.keys(appendedOp.vectorClock).length).toBeGreaterThan( MAX_VECTOR_CLOCK_SIZE, ); expect(appendedOp.vectorClock[TEST_CLIENT_ID]).toBeDefined(); }); - it('should preserve protected client IDs during pruning of local-win ops', async () => { + it('should preserve all client IDs in unpruned local-win op clock', async () => { const protectedId = 'protected-sync-import-client'; - mockOpLogStore.getProtectedClientIds.and.resolveTo([protectedId]); const now = Date.now(); - // Protected client has the lowest counter, should still be preserved const localClock: VectorClock = { [protectedId]: 1 }; for (let i = 1; i <= 5; i++) { localClock[`high-local-${i}`] = i * 100; @@ -2404,11 +2401,9 @@ describe('ConflictResolutionService', () => { const appendedOp = mockOpLogStore.appendWithVectorClockUpdate.calls.first() .args[0] as Operation; + // No client-side pruning — all keys preserved including protected client expect(appendedOp.vectorClock[protectedId]).toBeDefined(); expect(appendedOp.vectorClock[TEST_CLIENT_ID]).toBeDefined(); - expect(Object.keys(appendedOp.vectorClock).length).toBeLessThanOrEqual( - MAX_VECTOR_CLOCK_SIZE, - ); }); }); diff --git a/src/app/op-log/sync/conflict-resolution.service.ts b/src/app/op-log/sync/conflict-resolution.service.ts index 1cf9e7865f..08ef9ebf67 100644 --- a/src/app/op-log/sync/conflict-resolution.service.ts +++ b/src/app/op-log/sync/conflict-resolution.service.ts @@ -22,7 +22,6 @@ import { MAX_CONFLICT_RETRY_ATTEMPTS } from '../core/operation-log.const'; import { compareVectorClocks, incrementVectorClock, - limitVectorClockSize, mergeVectorClocks, VectorClockComparison, } from '../../core/util/vector-clock'; @@ -738,11 +737,10 @@ export class ConflictResolutionService { ...conflict.localOps.map((op) => op.vectorClock), ...conflict.remoteOps.map((op) => op.vectorClock), ]; - const mergedClock = this.mergeAndIncrementClocks(allClocks, clientId); - // Prune to match what the regular capture pipeline produces, preventing - // infinite rejection loops when server compares pruned vs unpruned clocks - const protectedClientIds = await this.opLogStore.getProtectedClientIds(); - const newClock = limitVectorClockSize(mergedClock, clientId, protectedClientIds); + // No client-side pruning — server prunes AFTER conflict detection, BEFORE storage. + // Client-side pruning can drop entity clock IDs, causing the server's pruning-aware + // comparison to return CONCURRENT instead of GREATER_THAN (infinite rejection loop). + const newClock = this.mergeAndIncrementClocks(allClocks, clientId); // Preserve the maximum timestamp from local ops. // This is critical for LWW semantics: we're creating a new op to carry the @@ -782,11 +780,8 @@ export class ConflictResolutionService { ...conflict.localOps.map((op) => op.vectorClock), ...conflict.remoteOps.map((op) => op.vectorClock), ]; - const mergedClock = this.mergeAndIncrementClocks(allClocks, clientId); - // Prune to match what the regular capture pipeline produces, preventing - // infinite rejection loops when server compares pruned vs unpruned clocks - const protectedClientIds = await this.opLogStore.getProtectedClientIds(); - const newClock = limitVectorClockSize(mergedClock, clientId, protectedClientIds); + // No client-side pruning — server prunes AFTER conflict detection, BEFORE storage. + const newClock = this.mergeAndIncrementClocks(allClocks, clientId); return { id: uuidv7(), diff --git a/src/app/op-log/sync/rejected-ops-handler.service.ts b/src/app/op-log/sync/rejected-ops-handler.service.ts index f1bf96e896..659e092f9d 100644 --- a/src/app/op-log/sync/rejected-ops-handler.service.ts +++ b/src/app/op-log/sync/rejected-ops-handler.service.ts @@ -82,9 +82,7 @@ export class RejectedOpsHandlerService { ): Promise { if (rejectedOps.length === 0) { // No rejections = sync is healthy, reset resolution attempt counters - if (this._resolutionAttemptsByEntity.size > 0) { - this._resolutionAttemptsByEntity.clear(); - } + this._resolutionAttemptsByEntity.clear(); return { mergedOpsCreated: 0, permanentRejectionCount: 0 }; } @@ -179,28 +177,27 @@ export class RejectedOpsHandlerService { OpLog.normal( `RejectedOpsHandlerService: Marked ${permanentlyRejectedOps.length} server-rejected ops as rejected`, ); - - // Notify user when any ops are permanently rejected - if (permanentlyRejectedOps.length > 0) { - this.snackService.open({ - type: 'ERROR', - msg: T.F.SYNC.S.UPLOAD_OPS_REJECTED, - translateParams: { count: permanentlyRejectedOps.length }, - }); - } + this.snackService.open({ + type: 'ERROR', + msg: T.F.SYNC.S.UPLOAD_OPS_REJECTED, + translateParams: { count: permanentlyRejectedOps.length }, + }); } // For concurrent modifications: try download first, then resolve locally if needed + let retryExceededCount = 0; if (concurrentModificationOps.length > 0 && downloadCallback) { - mergedOpsCreated = await this._resolveConcurrentModifications( + const result = await this._resolveConcurrentModifications( concurrentModificationOps, downloadCallback, ); + mergedOpsCreated = result.mergedOpsCreated; + retryExceededCount = result.retryExceededCount; } return { mergedOpsCreated, - permanentRejectionCount: permanentlyRejectedOps.length, + permanentRejectionCount: permanentlyRejectedOps.length + retryExceededCount, }; } @@ -214,7 +211,7 @@ export class RejectedOpsHandlerService { existingClock?: VectorClock; }>, downloadCallback: DownloadCallback, - ): Promise { + ): Promise<{ mergedOpsCreated: number; retryExceededCount: number }> { let mergedOpsCreated = 0; // Check resolution attempt counts per entity to prevent infinite loops. @@ -233,8 +230,7 @@ export class RejectedOpsHandlerService { }> = []; for (const item of concurrentModificationOps) { - const entityId = item.op.entityId || item.op.entityIds?.[0] || '*'; - const entityKey = `${item.op.entityType}:${entityId}`; + const entityKey = this._getEntityKey(item.op); const attempts = (this._resolutionAttemptsByEntity.get(entityKey) ?? 0) + 1; this._resolutionAttemptsByEntity.set(entityKey, attempts); @@ -259,14 +255,12 @@ export class RejectedOpsHandlerService { }); // Clean up tracking for rejected entities for (const item of opsExceededRetries) { - const entityId = item.op.entityId || item.op.entityIds?.[0] || '*'; - const entityKey = `${item.op.entityType}:${entityId}`; - this._resolutionAttemptsByEntity.delete(entityKey); + this._resolutionAttemptsByEntity.delete(this._getEntityKey(item.op)); } } if (opsToResolve.length === 0) { - return 0; + return { mergedOpsCreated: 0, retryExceededCount: opsExceededRetries.length }; } OpLog.normal( @@ -412,6 +406,11 @@ export class RejectedOpsHandlerService { throw e; } - return mergedOpsCreated; + return { mergedOpsCreated, retryExceededCount: opsExceededRetries.length }; + } + + private _getEntityKey(op: Operation): string { + const entityId = op.entityId || op.entityIds?.[0] || '*'; + return `${op.entityType}:${entityId}`; } } diff --git a/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts b/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts index 56175b3199..2b3ce31d6d 100644 --- a/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts +++ b/src/app/op-log/sync/superseded-operation-resolver.service.spec.ts @@ -45,7 +45,6 @@ describe('SupersededOperationResolverService', () => { mockOpLogStore = jasmine.createSpyObj('OperationLogStoreService', [ 'markRejected', 'appendWithVectorClockUpdate', - 'getProtectedClientIds', ]); mockVectorClockService = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', @@ -67,7 +66,6 @@ describe('SupersededOperationResolverService', () => { mockVectorClockService.getCurrentVectorClock.and.returnValue(Promise.resolve({})); mockOpLogStore.markRejected.and.returnValue(Promise.resolve()); mockOpLogStore.appendWithVectorClockUpdate.and.returnValue(Promise.resolve(1)); - mockOpLogStore.getProtectedClientIds.and.returnValue(Promise.resolve([])); // Mock lock service to execute the callback immediately mockLockService.request.and.callFake( (_lockName: string, callback: () => Promise) => callback(), @@ -1215,9 +1213,6 @@ describe('SupersededOperationResolverService', () => { it('should include all client IDs in unpruned merged clock', async () => { const protectedId = 'protected-sync-import-client'; - mockOpLogStore.getProtectedClientIds.and.returnValue( - Promise.resolve([protectedId]), - ); const globalClock: VectorClock = { [protectedId]: 1 }; for (let i = 1; i <= 10; i++) { @@ -1251,16 +1246,13 @@ describe('SupersededOperationResolverService', () => { it('should include existingClock client IDs in unpruned merged clock', async () => { // Simulate scenario where existingClock has server entity client IDs // that must be preserved for the server to see GREATER_THAN (not CONCURRENT). - const protectedIds = Array.from( + const clientIds = Array.from( { length: MAX_VECTOR_CLOCK_SIZE }, - (_, i) => `protected-${i}`, - ); - mockOpLogStore.getProtectedClientIds.and.returnValue( - Promise.resolve(protectedIds), + (_, i) => `client-${i}`, ); const globalClock: VectorClock = {}; - for (const id of protectedIds) { + for (const id of clientIds) { globalClock[id] = 5; } globalClock['serverEntityClient'] = 7; // Server entity clock entry @@ -1299,10 +1291,6 @@ describe('SupersededOperationResolverService', () => { { length: MAX_VECTOR_CLOCK_SIZE }, (_, i) => `protected-${i}`, ); - mockOpLogStore.getProtectedClientIds.and.returnValue( - Promise.resolve(protectedIds), - ); - const globalClock: VectorClock = {}; for (const id of protectedIds) { globalClock[id] = 5;