From 9328156ca1697170c6255552094026cb5b0cfc9f Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Sat, 14 Feb 2026 13:16:48 +0100 Subject: [PATCH] fix(sync): add vector clock pruning to compaction and update docs Add limitVectorClockSize to OperationLogCompactionService._doCompact() which was the remaining saveStateCache caller that persisted unpruned clocks. Update vector-clocks.md exhaustive pruning table with the three new client-side pruning locations. Add boundary tests at exactly MAX_VECTOR_CLOCK_SIZE for snapshot and hydrator services. --- docs/sync-and-op-log/vector-clocks.md | 3 ++ .../operation-log-compaction.service.spec.ts | 44 +++++++++++++++++++ .../operation-log-compaction.service.ts | 12 ++++- .../operation-log-hydrator.service.spec.ts | 13 ++++++ .../operation-log-snapshot.service.spec.ts | 16 +++++++ 5 files changed, 87 insertions(+), 1 deletion(-) diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index 1e41bf988c..6167892b0b 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -153,6 +153,9 @@ Implemented in `packages/shared-schema/src/vector-clock.ts`. The client wrapper | **Client** `SyncHydrationService` | Creating SYNC_IMPORT during conflict resolution | Current client only | | **Client** `ServerMigrationService` | Creating SYNC_IMPORT during migration | Current client only | | **Client** `RepairOperationService` | Creating REPAIR operation | Current client only | +| **Client** `OperationLogSnapshotService` | Saving snapshot to state cache | Current client only | +| **Client** `OperationLogCompactionService` | Compaction (saving snapshot + deleting old ops) | Current client only | +| **Client** `OperationLogHydratorService` | Restoring snapshot during hydration | Current client only | | **Client** normal op capture | **NEVER** | N/A | | **Client** `SupersededOperationResolverService` | **NEVER** (conflict resolution) | N/A | diff --git a/src/app/op-log/persistence/operation-log-compaction.service.spec.ts b/src/app/op-log/persistence/operation-log-compaction.service.spec.ts index f423c6cbd7..7b5b2ef526 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.spec.ts @@ -12,6 +12,8 @@ import { CURRENT_SCHEMA_VERSION } from './schema-migration.service'; import { OperationLogEntry } from '../core/operation.types'; import { OpLog } from '../../core/log'; import { MODEL_CONFIGS } from '../model/model-config'; +import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; +import { MAX_VECTOR_CLOCK_SIZE } from '@sp/shared-schema'; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -21,6 +23,7 @@ describe('OperationLogCompactionService', () => { let mockLockService: jasmine.SpyObj; let mockStateSnapshot: jasmine.SpyObj; let mockVectorClockService: jasmine.SpyObj; + let mockClientIdProvider: jasmine.SpyObj; const mockState = { task: { entities: {}, ids: [] }, @@ -44,6 +47,8 @@ describe('OperationLogCompactionService', () => { mockVectorClockService = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', ]); + mockClientIdProvider = jasmine.createSpyObj('ClientIdProvider', ['loadClientId']); + mockClientIdProvider.loadClientId.and.resolveTo('test-client'); // Mock OpLog methods spyOn(OpLog, 'warn'); @@ -71,6 +76,7 @@ describe('OperationLogCompactionService', () => { { provide: LockService, useValue: mockLockService }, { provide: StateSnapshotService, useValue: mockStateSnapshot }, { provide: VectorClockService, useValue: mockVectorClockService }, + { provide: CLIENT_ID_PROVIDER, useValue: mockClientIdProvider }, ], }); @@ -151,6 +157,44 @@ describe('OperationLogCompactionService', () => { ); }); + it('should prune vector clock before saving when it exceeds MAX_VECTOR_CLOCK_SIZE', async () => { + const bloatedClock: Record = {}; + for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 10; i++) { + bloatedClock[`client-${i}`] = i + 1; + } + bloatedClock['test-client'] = 999; + mockVectorClockService.getCurrentVectorClock.and.resolveTo(bloatedClock); + + await service.compact(); + + const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; + expect(Object.keys(savedCache.vectorClock).length).toBeLessThanOrEqual( + MAX_VECTOR_CLOCK_SIZE, + ); + expect(savedCache.vectorClock['test-client']).toBe(999); + }); + + it('should not prune vector clock when within MAX_VECTOR_CLOCK_SIZE', async () => { + const smallClock = { clientA: 10, clientB: 5 }; + mockVectorClockService.getCurrentVectorClock.and.resolveTo(smallClock); + + await service.compact(); + + const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; + expect(savedCache.vectorClock).toEqual(smallClock); + }); + + it('should save unpruned clock if clientId is null', async () => { + mockClientIdProvider.loadClientId.and.resolveTo(null); + const clock = { clientA: 10 }; + mockVectorClockService.getCurrentVectorClock.and.resolveTo(clock); + + await service.compact(); + + const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; + expect(savedCache.vectorClock).toEqual(clock); + }); + it('should save state cache with compactedAt timestamp', async () => { const beforeTime = Date.now(); await service.compact(); diff --git a/src/app/op-log/persistence/operation-log-compaction.service.ts b/src/app/op-log/persistence/operation-log-compaction.service.ts index 570a33733c..f04bcf2660 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.ts @@ -13,6 +13,8 @@ import { CURRENT_SCHEMA_VERSION } from './schema-migration.service'; import { VectorClockService } from '../sync/vector-clock.service'; import { OpLog } from '../../core/log'; import { extractEntityKeysFromState } from './extract-entity-keys'; +import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; +import { limitVectorClockSize } from '../../core/util/vector-clock'; /** * Manages the compaction (garbage collection) of the operation log. @@ -28,6 +30,7 @@ export class OperationLogCompactionService { private lockService = inject(LockService); private stateSnapshot = inject(StateSnapshotService); private vectorClockService = inject(VectorClockService); + private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); async compact(): Promise { await this._doCompact(COMPACTION_RETENTION_MS, false); @@ -66,6 +69,13 @@ export class OperationLogCompactionService { const currentVectorClock = await this.vectorClockService.getCurrentVectorClock(); this.checkCompactionTimeout(startTime, `${label}vector clock`); + // Prune vector clock before persisting to prevent bloat (max 20 entries). + // Without this, clocks can grow unbounded across sync cycles. + const clientId = await this.clientIdProvider.loadClientId(); + const prunedClock = clientId + ? limitVectorClockSize(currentVectorClock, clientId) + : currentVectorClock; + // 3. Get lastSeq IMMEDIATELY before writing cache to minimize race window // This ensures new ops written after this point have seq > lastSeq const lastSeq = await this.opLogStore.getLastSeq(); @@ -79,7 +89,7 @@ export class OperationLogCompactionService { await this.opLogStore.saveStateCache({ state: currentState, lastAppliedOpSeq: lastSeq, - vectorClock: currentVectorClock, + vectorClock: prunedClock, compactedAt: Date.now(), schemaVersion: CURRENT_SCHEMA_VERSION, snapshotEntityKeys, diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts index 7c463f8ce3..150dee9cac 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.spec.ts @@ -388,6 +388,19 @@ describe('OperationLogHydratorService', () => { expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith(smallClock); }); + it('should not prune vector clock at exactly MAX_VECTOR_CLOCK_SIZE entries', async () => { + const exactClock: Record = {}; + for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE; i++) { + exactClock[`client-${i}`] = i + 1; + } + const snapshot = createMockSnapshot({ vectorClock: exactClock }); + mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); + + await service.hydrateStore(); + + expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith(exactClock); + }); + it('should restore unpruned clock if clientId is null', async () => { mockClientIdProvider.loadClientId.and.resolveTo(null); const clock = { clientA: 5 }; diff --git a/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts b/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts index 7868fb80a2..457f88aa8f 100644 --- a/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-snapshot.service.spec.ts @@ -215,6 +215,22 @@ describe('OperationLogSnapshotService', () => { expect(savedCache.vectorClock).toEqual(smallClock); }); + it('should not prune vector clock at exactly MAX_VECTOR_CLOCK_SIZE entries', async () => { + const exactClock: Record = {}; + for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE; i++) { + exactClock[`client-${i}`] = i + 1; + } + mockStateSnapshotService.getStateSnapshot.and.returnValue({} as any); + mockVectorClockService.getCurrentVectorClock.and.resolveTo(exactClock); + mockOpLogStore.getLastSeq.and.resolveTo(1); + mockOpLogStore.saveStateCache.and.resolveTo(undefined); + + await service.saveCurrentStateAsSnapshot(); + + const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; + expect(savedCache.vectorClock).toEqual(exactClock); + }); + it('should save unpruned clock if clientId is null', async () => { mockClientIdProvider.loadClientId.and.resolveTo(null); const clock = { client1: 5 };