From fdbc2c1a86f4022b211017d0e93f90b1633cd410 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Fri, 17 Jul 2026 13:12:53 +0200 Subject: [PATCH] refactor(sync): centralize clock pruning in store, make merge atomic (#9107) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to f2b533af41 (#9096). Two changes: 1. Store-owned pruning choke point. The preserve-set invariant (current client + latest full-state author) was duplicated at five caller sites — exactly how #9089/#9096 happened. OperationLogStoreService now owns it in pruneClockForStorage; setVectorClock, saveStateCache, and commitFileSnapshotBaseline prune internally, and callers (snapshot, compaction, hydrator, sync-hydration, server-migration) pass raw clocks. This also covers the previously unpruned setVectorClock site in _restorePreservedLocalOps. Importing limitVectorClockSize outside the store (client wrapper or @sp/sync-core) now fails lint (no-restricted-imports, sabotage-verified on both import routes). 2. Atomic mergeRemoteOpClocks. The merge was a read-compute-put across separate transactions reading the per-tab cache — two tabs could interleave and lose clock entries. It is now one readwrite transaction with a fresh in-transaction read of the durable clock, mirroring the reducer checkpoint (regression test proves the stale- cache lost update). Pruning behavior tests move to the store spec (caller specs mock the store and can no longer observe pruning); caller-level prune tests are deleted, the sync-hydration case becomes a wiring test. Docs updated. --- docs/sync-and-op-log/vector-clocks.md | 28 ++-- eslint.config.js | 31 ++++ .../operation-log-compaction.service.spec.ts | 59 +------- .../operation-log-compaction.service.ts | 22 +-- .../operation-log-hydrator.service.spec.ts | 58 +------- .../operation-log-hydrator.service.ts | 20 +-- .../operation-log-snapshot.service.spec.ts | 41 +----- .../operation-log-snapshot.service.ts | 23 +-- .../operation-log-store.service.spec.ts | 131 ++++++++++++++--- .../operation-log-store.service.ts | 139 ++++++++++++------ .../sync-hydration.service.spec.ts | 38 ++--- .../persistence/sync-hydration.service.ts | 21 +-- .../sync/server-migration.service.spec.ts | 3 + .../op-log/sync/server-migration.service.ts | 18 +-- .../post-sync-validation.integration.spec.ts | 7 +- 15 files changed, 312 insertions(+), 327 deletions(-) diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index 217cc977c9..fcef56bdda 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -77,7 +77,7 @@ interface VectorClockEntry { } ``` -The global clock is the **single source of truth** for the client's current causal knowledge. During local operation capture, it is updated atomically with operation writes (single IndexedDB transaction via `appendWithVectorClockOverwrite`). The remote merge path (`mergeRemoteOpClocks`) updates the clock in a separate write after reading the current state. +The global clock is the **single source of truth** for the client's current causal knowledge. During local operation capture, it is updated atomically with operation writes (single IndexedDB transaction via `appendWithVectorClockOverwrite`). The remote merge path (`mergeRemoteOpClocks`) is likewise a single read-merge-write transaction with a fresh in-transaction read of the durable clock — never the per-tab cache — so concurrent tabs cannot lose entries to a stale read. ### Snapshot Clock @@ -142,23 +142,21 @@ Otherwise: 3. Return clock with exactly MAX entries ``` -Implemented in `packages/sync-core/src/vector-clock.ts`. The client wrapper in `src/app/core/util/vector-clock.ts` adds logging and takes a caller-supplied preserve list — the current client plus, where a full-state baseline exists, the latest full-state author (#9096). +Implemented in `packages/sync-core/src/vector-clock.ts`. Client-side pruning is **store-owned** (#9096): `OperationLogStoreService.pruneClockForStorage` assembles the preserve set — current client + latest full-state author — and every durable-clock write routes through it. Importing `limitVectorClockSize` anywhere else in `src/app` fails lint (`no-restricted-imports`); the wrapper in `src/app/core/util/vector-clock.ts` (adds logging) is importable only by the store. ### When Pruning Happens (Exhaustive List) -| Location | When | What's Preserved | -| ------------------------------------------------------------------- | ----------------------------------------------------------------------------------- | ------------------------------------------- | -| **Server** `processOperation()` | After conflict detection, before storage | Uploading client + active full-state author | -| **Server** `getOpsSinceWithSeq()` | Aggregating snapshot vector clock | Requesting client | -| **Client** `calculateRemoteClockMerge` (`OperationLogStoreService`) | Durable clock after a remote batch (merge + reducer checkpoint) | Current client + latest full-state author | -| **Client** `SyncHydrationService` | Creating SYNC_IMPORT / committing file-snapshot bootstrap baseline | Current client + latest full-state author | -| **Client** `ServerMigrationService` | Creating SYNC_IMPORT during migration | Current client (the new import author) | -| **Client** `OperationLogSnapshotService` | Saving snapshot to state cache | Current client + latest full-state author | -| **Client** `OperationLogCompactionService` | Compaction (saving snapshot + deleting old ops) | Current client + latest full-state author | -| **Client** `OperationLogHydratorService` | Restoring snapshot clock into the durable clock at hydration | Current client + latest full-state author | -| **Client** `RepairOperationService` | **NEVER** — REPAIR ships the full clock; the server prunes after conflict detection | N/A | -| **Client** normal op capture | **NEVER** | N/A | -| **Client** `SupersededOperationResolverService` | **NEVER** (conflict resolution) | N/A | +| Location | When | What's Preserved | +| ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------- | ------------------------------------------- | +| **Server** `processOperation()` | After conflict detection, before storage | Uploading client + active full-state author | +| **Server** `getOpsSinceWithSeq()` | Aggregating snapshot vector clock | Requesting client | +| **Client** `OperationLogStoreService` — `calculateRemoteClockMerge` (remote merge + reducer checkpoint, in-transaction) | Durable clock after a remote batch | Current client + latest full-state author | +| **Client** `OperationLogStoreService.pruneClockForStorage` — inside `setVectorClock`, `saveStateCache`, `commitFileSnapshotBaseline`; called directly by `SyncHydrationService` / `ServerMigrationService` for SYNC_IMPORT op clocks | Every other durable-clock write (snapshot save, compaction, hydration restore, sync-hydration baseline, imports) | Current client + latest full-state author | +| **Client** callers (snapshot, compaction, hydrator, sync-hydration, server-migration) | **NEVER** — they pass raw clocks; the store prunes (lint-enforced) | N/A | +| **Client** in-store direct clock writes (`appendWithVectorClockOverwrite`, `runRemoteStateReplacement`, `runDestructiveStateReplacement`, `appendRecoveryOperationAndSnapshot`) | **NEVER** — write full, minimal, or already-server-pruned clocks by design | N/A | +| **Client** `RepairOperationService` | **NEVER** — REPAIR ships the full clock; the server prunes after conflict detection | N/A | +| **Client** normal op capture | **NEVER** | N/A | +| **Client** `SupersededOperationResolverService` | **NEVER** (conflict resolution) | N/A | ### Pruning is Rare diff --git a/eslint.config.js b/eslint.config.js index 97df05e0b7..d5628d0578 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -263,6 +263,37 @@ module.exports = tseslint.config( 'no-console': 'error', }, }, + // Durable-clock pruning is store-owned (#9096): every clock persisted by + // the client must be pruned with the full preserve set (current client + + // latest full-state author), which OperationLogStoreService assembles in + // pruneClockForStorage. Caller-site pruning is how the import author got + // silently evicted (#9089/#9096), so importing limitVectorClockSize outside + // the store (from the client wrapper or @sp/sync-core) is fenced off. + // Exempt: the wrapper itself (re-exports the shared impl), the store + // service (the choke point), and specs (simulate server-side pruning). + { + files: ['src/app/**/*.ts'], + ignores: [ + 'src/app/**/*.spec.ts', + 'src/app/core/util/vector-clock.ts', + 'src/app/op-log/persistence/operation-log-store.service.ts', + ], + rules: { + 'no-restricted-imports': [ + 'error', + { + patterns: [ + { + group: ['**/core/util/vector-clock', '@sp/sync-core'], + importNamePattern: '^limitVectorClockSize$', + message: + 'Durable-clock pruning is store-owned (#9096): use OperationLogStoreService.pruneClockForStorage instead of pruning at the call site.', + }, + ], + }, + ], + }, + }, // Service size cap (AGENTS.md → Project rules): no service may exceed 1200 // lines. 'error' so a new service crossing the cap fails CI on the PR that // introduces it — 'warn' would be inert, since `ng lint` defaults to 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 7c6f7ac1b9..75a8fad3d8 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 @@ -6,11 +6,10 @@ import { StateSnapshotService } from '../backup/state-snapshot.service'; import { VectorClockService } from '../sync/vector-clock.service'; import { COMPACTION_RETENTION_MS, - MAX_VECTOR_CLOCK_SIZE, SLOW_COMPACTION_THRESHOLD_MS, } from '../core/operation-log.const'; import { CURRENT_SCHEMA_VERSION } from './schema-migration.service'; -import { Operation, OperationLogEntry, OpType } from '../core/operation.types'; +import { OperationLogEntry, OpType } 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'; @@ -57,9 +56,7 @@ describe('OperationLogCompactionService', () => { 'resetCompactionCounter', 'deleteOpsWhere', 'getPendingRemoteOps', - 'getLatestFullStateOp', ]); - mockOpLogStore.getLatestFullStateOp.and.resolveTo(undefined); mockLockService = jasmine.createSpyObj('LockService', ['request']); mockStateSnapshot = jasmine.createSpyObj('StateSnapshotService', [ 'getStateSnapshot', @@ -196,45 +193,10 @@ 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 keep the latest import author when pruning the snapshot clock (#9096)', async () => { - // The author's counter is the lowest, so it survives only through the - // preserve list; this clock becomes the durable clock at hydration. - const bloatedClock: Record = { importAuthor: 1 }; - for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 10; i++) { - bloatedClock[`client-${i}`] = i + 100; - } - bloatedClock['test-client'] = 999; - mockVectorClockService.getCurrentVectorClock.and.resolveTo(bloatedClock); - mockOpLogStore.getLatestFullStateOp.and.resolveTo({ - clientId: 'importAuthor', - } as Operation); - - await service.compact(); - - const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; - expect(Object.keys(savedCache.vectorClock).length).toBe(MAX_VECTOR_CLOCK_SIZE); - expect(savedCache.vectorClock['importAuthor']).toBe(1); - expect(savedCache.vectorClock['test-client']).toBe(999); - }); - - it('should not prune vector clock when within MAX_VECTOR_CLOCK_SIZE', async () => { + // Vector-clock pruning is store-owned (saveStateCache prunes internally, + // #9096) — covered by the OperationLogStoreService spec. This service + // passes the clock through unmodified: + it('should pass the current vector clock through to saveStateCache unmodified', async () => { const smallClock = { clientA: 10, clientB: 5 }; mockVectorClockService.getCurrentVectorClock.and.resolveTo(smallClock); @@ -244,17 +206,6 @@ describe('OperationLogCompactionService', () => { 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 40b76bbae5..b3024f8c3d 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.ts @@ -13,8 +13,6 @@ 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'; import { hasMeaningfulStateData } from '../validation/has-meaningful-state-data.util'; import { OperationCaptureService } from '../capture/operation-capture.service'; import { getPhantomChangeRisk } from '../capture/phantom-change-guard.util'; @@ -34,7 +32,6 @@ export class OperationLogCompactionService { private lockService = inject(LockService); private stateSnapshot = inject(StateSnapshotService); private vectorClockService = inject(VectorClockService); - private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); private operationCapture = inject(OperationCaptureService); private writeFlushService = inject(OperationWriteFlushService); @@ -144,24 +141,11 @@ export class OperationLogCompactionService { return false; } - // 2. Get current vector clock (max of all ops) + // 2. Get current vector clock (max of all ops); pruning happens inside + // saveStateCache (store-owned, #9096) 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. The latest - // full-state author is preserved alongside the current client — this - // clock is restored as the durable clock at hydration, so evicting the - // author would make later ops CONCURRENT with the import on peers (#9096). - const clientId = await this.clientIdProvider.loadClientId(); - const importAuthorId = (await this.opLogStore.getLatestFullStateOp())?.clientId; - const prunedClock = clientId - ? limitVectorClockSize( - currentVectorClock, - importAuthorId ? [clientId, importAuthorId] : [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(); @@ -175,7 +159,7 @@ export class OperationLogCompactionService { await this.opLogStore.saveStateCache({ state: currentState, lastAppliedOpSeq: lastSeq, - vectorClock: prunedClock, + vectorClock: currentVectorClock, 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 47cf8a7c8c..aaec345fe2 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 @@ -337,50 +337,9 @@ describe('OperationLogHydratorService', () => { expect(mockOpLogStore.setVectorClock).not.toHaveBeenCalled(); }); - it('should prune bloated vector clock before restoring from snapshot', async () => { - // Create a bloated vector clock with more entries than MAX_VECTOR_CLOCK_SIZE - const bloatedClock: Record = {}; - for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 10; i++) { - bloatedClock[`client-${i}`] = i + 1; - } - bloatedClock['test-client'] = 999; - - const snapshot = createMockSnapshot({ vectorClock: bloatedClock }); - mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); - - await service.hydrateStore(); - - const restoredClock = mockOpLogStore.setVectorClock.calls.mostRecent().args[0]; - expect(Object.keys(restoredClock).length).toBeLessThanOrEqual( - MAX_VECTOR_CLOCK_SIZE, - ); - // Local client ID must be preserved after pruning - expect(restoredClock['test-client']).toBe(999); - }); - - it('should keep the latest import author when pruning the restored clock (#9096)', async () => { - // The restored clock becomes the DURABLE clock; evicting the import - // author here would make every later op CONCURRENT with the import. - const bloatedClock: Record = { importAuthor: 1 }; - for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 10; i++) { - bloatedClock[`client-${i}`] = i + 100; - } - bloatedClock['test-client'] = 999; - - const snapshot = createMockSnapshot({ vectorClock: bloatedClock }); - mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); - mockOpLogStore.getLatestFullStateOp.and.returnValue( - Promise.resolve({ clientId: 'importAuthor' } as Operation), - ); - - await service.hydrateStore(); - - const restoredClock = mockOpLogStore.setVectorClock.calls.mostRecent().args[0]; - expect(Object.keys(restoredClock).length).toBe(MAX_VECTOR_CLOCK_SIZE); - expect(restoredClock['importAuthor']).toBe(1); - expect(restoredClock['test-client']).toBe(999); - }); - + // Vector-clock pruning is store-owned (setVectorClock prunes + // internally, #9096) — covered by the OperationLogStoreService spec. + // The hydrator passes the snapshot clock through unmodified: it('should not prune vector clock when within MAX_VECTOR_CLOCK_SIZE', async () => { const smallClock = { clientA: 5, clientB: 3 }; const snapshot = createMockSnapshot({ vectorClock: smallClock }); @@ -403,17 +362,6 @@ describe('OperationLogHydratorService', () => { expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith(exactClock); }); - - it('should restore unpruned clock if clientId is null', async () => { - mockClientIdProvider.loadClientId.and.resolveTo(null); - const clock = { clientA: 5 }; - const snapshot = createMockSnapshot({ vectorClock: clock }); - mockOpLogStore.loadStateCache.and.returnValue(Promise.resolve(snapshot)); - - await service.hydrateStore(); - - expect(mockOpLogStore.setVectorClock).toHaveBeenCalledWith(clock); - }); }); describe('tail operation replay', () => { diff --git a/src/app/op-log/persistence/operation-log-hydrator.service.ts b/src/app/op-log/persistence/operation-log-hydrator.service.ts index 8355bbf8c6..9df2857991 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.ts @@ -33,7 +33,6 @@ import { bulkApplyOperations } from '../apply/bulk-hydration.action'; import { VectorClockService } from '../sync/vector-clock.service'; import { AppDataComplete } from '../model/model-config'; import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; -import { limitVectorClockSize } from '../../core/util/vector-clock'; import { IS_ELECTRON } from '../../app.constants'; import { BulkReplayReducerFailure, @@ -205,23 +204,12 @@ export class OperationLogHydratorService { // 3. Without this, new ops would have clocks missing entries from the SYNC_IMPORT // 4. Those ops would be CONCURRENT with the SYNC_IMPORT and get filtered on sync if (snapshot.vectorClock && Object.keys(snapshot.vectorClock).length > 0) { - // Prune vector clock before restoring to prevent bloat from old snapshots - // that were saved before pruning was added to saveCurrentStateAsSnapshot(). - // Preserve the latest full-state author alongside the current client: - // this write becomes the durable clock, and evicting the author would - // make every later op CONCURRENT with the import on peers (#9096). - const clientId = await this.clientIdProvider.loadClientId(); - const importAuthorId = (await this.opLogStore.getLatestFullStateOp())?.clientId; - const clockToRestore = clientId - ? limitVectorClockSize( - snapshot.vectorClock, - importAuthorId ? [clientId, importAuthorId] : [clientId], - ) - : snapshot.vectorClock; - await this.opLogStore.setVectorClock(clockToRestore); + // setVectorClock prunes internally (store-owned, #9096) — this also + // bounds legacy snapshot clocks saved before pruning existed. + await this.opLogStore.setVectorClock(snapshot.vectorClock); OpLog.normal( 'OperationLogHydratorService: Restored vector clock from snapshot', - { clockSize: Object.keys(clockToRestore).length }, + { clockSize: Object.keys(snapshot.vectorClock).length }, ); } 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 85002c173b..de40323c99 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 @@ -55,9 +55,7 @@ describe('OperationLogSnapshotService', () => { 'clearStateCacheBackup', 'restoreStateCacheFromBackup', 'getLastSeq', - 'getLatestFullStateOp', ]); - mockOpLogStore.getLatestFullStateOp.and.resolveTo(undefined); mockVectorClockService = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', ]); @@ -266,31 +264,9 @@ describe('OperationLogSnapshotService', () => { expect(mockOpLogStore.saveStateCache).not.toHaveBeenCalled(); }); - it('should prune vector clock before saving when it exceeds MAX_VECTOR_CLOCK_SIZE', async () => { - // Create a bloated vector clock with more entries than MAX_VECTOR_CLOCK_SIZE - const bloatedClock: Record = {}; - for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 10; i++) { - bloatedClock[`client-${i}`] = i + 1; - } - // Ensure the local client is in the clock - bloatedClock['test-client'] = 999; - - mockStateSnapshotService.getStateSnapshot.and.returnValue( - MEANINGFUL_SNAPSHOT_STATE as any, - ); - mockVectorClockService.getCurrentVectorClock.and.resolveTo(bloatedClock); - mockOpLogStore.getLastSeq.and.resolveTo(1); - mockOpLogStore.saveStateCache.and.resolveTo(undefined); - - await service.saveCurrentStateAsSnapshot(); - - const savedCache = mockOpLogStore.saveStateCache.calls.mostRecent().args[0]; - const savedClockSize = Object.keys(savedCache.vectorClock).length; - expect(savedClockSize).toBeLessThanOrEqual(MAX_VECTOR_CLOCK_SIZE); - // Local client ID must be preserved after pruning - expect(savedCache.vectorClock['test-client']).toBe(999); - }); - + // Vector-clock pruning is store-owned (saveStateCache prunes internally, + // #9096) — covered by the OperationLogStoreService spec. This service + // passes the clock through unmodified: it('should not prune vector clock when it is within MAX_VECTOR_CLOCK_SIZE', async () => { const smallClock = { client1: 5, client2: 3 }; mockStateSnapshotService.getStateSnapshot.and.returnValue( @@ -324,8 +300,7 @@ describe('OperationLogSnapshotService', () => { expect(savedCache.vectorClock).toEqual(exactClock); }); - it('should save unpruned clock if clientId is null', async () => { - mockClientIdProvider.loadClientId.and.resolveTo(null); + it('should save the clock from the vector clock service verbatim', async () => { const clock = { client1: 5 }; mockStateSnapshotService.getStateSnapshot.and.returnValue( MEANINGFUL_SNAPSHOT_STATE as any, @@ -379,10 +354,10 @@ describe('OperationLogSnapshotService', () => { mockStateSnapshotService.getStateSnapshot.and.returnValue( MEANINGFUL_SNAPSHOT_STATE as any, ); - // Must be stubbed for the whole describe, not per test: without it the - // save path throws (limitVectorClockSize on undefined) into the outer - // catch and skips the write anyway — which makes every "should skip" - // test below pass even when the guard is deleted. + // Stubbed for the whole describe so the save path runs with a realistic + // clock. (Historically load-bearing: the service used to prune here and + // threw on an unstubbed clock, which made the "should skip" tests below + // vacuously green; pruning is store-owned now — #9096.) mockVectorClockService.getCurrentVectorClock.and.resolveTo({ c1: 1 }); clearDeferredActions(); }); diff --git a/src/app/op-log/persistence/operation-log-snapshot.service.ts b/src/app/op-log/persistence/operation-log-snapshot.service.ts index aea88065f7..4e847f5767 100644 --- a/src/app/op-log/persistence/operation-log-snapshot.service.ts +++ b/src/app/op-log/persistence/operation-log-snapshot.service.ts @@ -9,8 +9,6 @@ import { VectorClockService } from '../sync/vector-clock.service'; import { StateSnapshotService } from '../backup/state-snapshot.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'; import { ValidateStateService } from '../validation/validate-state.service'; import { hasMeaningfulStateData } from '../validation/has-meaningful-state-data.util'; import { OperationCaptureService } from '../capture/operation-capture.service'; @@ -37,7 +35,6 @@ export class OperationLogSnapshotService { private stateSnapshotService = inject(StateSnapshotService); private schemaMigrationService = inject(SchemaMigrationService); private validateStateService = inject(ValidateStateService); - private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER); private operationCapture = inject(OperationCaptureService); private writeFlushService = inject(OperationWriteFlushService); @@ -149,24 +146,10 @@ export class OperationLogSnapshotService { return; } - // Get current vector clock + // Get current vector clock; pruning happens inside saveStateCache + // (store-owned, #9096). const vectorClock = await this.vectorClockService.getCurrentVectorClock(); - // Prune vector clock before persisting to prevent bloat (max 20 entries). - // Without this, clocks can grow unbounded across sync cycles and cause - // repeated conflict dialogs on every sync. The latest full-state author - // is preserved alongside the current client — the hydrator restores - // this clock as the durable clock, so evicting the author here would - // make every later op CONCURRENT with the import on peers (#9096). - const clientId = await this.clientIdProvider.loadClientId(); - const importAuthorId = (await this.opLogStore.getLatestFullStateOp())?.clientId; - const prunedClock = clientId - ? limitVectorClockSize( - vectorClock, - importAuthorId ? [clientId, importAuthorId] : [clientId], - ) - : vectorClock; - // Extract entity keys for conflict detection after compaction const snapshotEntityKeys = extractEntityKeysFromState(currentState); @@ -174,7 +157,7 @@ export class OperationLogSnapshotService { await this.opLogStore.saveStateCache({ state: currentState, lastAppliedOpSeq: lastSeq, - vectorClock: prunedClock, + vectorClock, compactedAt: Date.now(), schemaVersion: CURRENT_SCHEMA_VERSION, snapshotEntityKeys, diff --git a/src/app/op-log/persistence/operation-log-store.service.spec.ts b/src/app/op-log/persistence/operation-log-store.service.spec.ts index 8ad2865283..88c0f69033 100644 --- a/src/app/op-log/persistence/operation-log-store.service.spec.ts +++ b/src/app/op-log/persistence/operation-log-store.service.spec.ts @@ -65,6 +65,33 @@ describe('OperationLogStoreService', () => { ...overrides, }); + const createImportOp = (clientId: string, counter: number): Operation => + createTestOperation({ + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + clientId, + vectorClock: { [clientId]: counter }, + }); + + const createBusyClientOps = (count: number): Operation[] => + Array.from({ length: count }, (_, i) => + createTestOperation({ + clientId: `busyClient_${i}`, + vectorClock: { [`busyClient_${i}`]: 100 + i }, + }), + ); + + // A >MAX clock whose lowest counters belong to the ids under test — the + // shape where uploader/author protection is load-bearing (#9096). + const createBloatedClock = (lowCounterEntries: VectorClock): VectorClock => { + const clock: VectorClock = { ...lowCounterEntries }; + for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 5; i++) { + clock[`bloatClient_${i}`] = 100 + i; + } + return clock; + }; + beforeEach(async () => { TestBed.configureTestingModule({ providers: [ @@ -3966,23 +3993,6 @@ describe('OperationLogStoreService', () => { }); describe('import author protection during clock pruning (#9096)', () => { - const createImportOp = (clientId: string, counter: number): Operation => - createTestOperation({ - opType: OpType.SyncImport, - entityType: 'ALL' as EntityType, - entityId: undefined, - clientId, - vectorClock: { [clientId]: counter }, - }); - - const createBusyClientOps = (count: number): Operation[] => - Array.from({ length: count }, (_, i) => - createTestOperation({ - clientId: `busyClient_${i}`, - vectorClock: { [`busyClient_${i}`]: 100 + i }, - }), - ); - it('should keep the stored import author when a remote batch overflows the clock', async () => { // Durable baseline after receiving an import from another client: the // author's counter is LOW, so uploader-only pruning would evict it first. @@ -4055,6 +4065,93 @@ describe('OperationLogStoreService', () => { }); }); + describe('store-owned durable-clock pruning (pruneClockForStorage)', () => { + it('should prune an over-MAX clock in setVectorClock, keeping the latest import author', async () => { + await service.append(createImportOp('importAuthor', 1), 'remote'); + + await service.setVectorClock( + createBloatedClock({ importAuthor: 1, testClient: 999 }), + ); + + const clock = await service.getVectorClock(); + expect(Object.keys(clock!).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock!['importAuthor']).toBe(1); + expect(clock!['testClient']).toBe(999); + }); + + it('should preserve only the current client in setVectorClock when no full-state baseline exists', async () => { + // Shape of the USE_REMOTE raw-rebuild resume (_restorePreservedLocalOps): + // ops store cleared, no import — self must survive on its own. + await service.setVectorClock(createBloatedClock({ testClient: 1 })); + + const clock = await service.getVectorClock(); + expect(Object.keys(clock!).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock!['testClient']).toBe(1); + }); + + it('should prune the snapshot clock in saveStateCache, keeping the latest import author', async () => { + await service.append(createImportOp('importAuthor', 1), 'remote'); + + await service.saveStateCache({ + state: { some: 'state' }, + lastAppliedOpSeq: 1, + vectorClock: createBloatedClock({ importAuthor: 1, testClient: 999 }), + compactedAt: Date.now(), + }); + + const cache = await service.loadStateCache(); + expect(Object.keys(cache!.vectorClock).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(cache!.vectorClock['importAuthor']).toBe(1); + expect(cache!.vectorClock['testClient']).toBe(999); + }); + + it('should prune the baseline clock in commitFileSnapshotBaseline for cache AND durable clock', async () => { + await service.append(createImportOp('importAuthor', 1), 'remote'); + const lastSeq = await service.getLastSeq(); + + await service.commitFileSnapshotBaseline({ + state: { some: 'state' }, + lastAppliedOpSeq: lastSeq, + vectorClock: createBloatedClock({ importAuthor: 1, testClient: 999 }), + compactedAt: Date.now(), + snapshotIncludedOps: [], + }); + + const cacheClock = (await service.loadStateCache())!.vectorClock; + const durableClock = (await service.getVectorClock())!; + for (const clock of [cacheClock, durableClock]) { + expect(Object.keys(clock).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock['importAuthor']).toBe(1); + expect(clock['testClient']).toBe(999); + } + }); + + it('should merge onto the durable clock, not the per-tab cache (multi-tab lost update)', async () => { + await service.setVectorClock({ a: 1 }); + await service.getVectorClock(); // populate this tab's in-memory cache + + // Another tab advances the durable clock behind this tab's cache. + await ( + service as unknown as { + _adapter: { + put: (store: string, value: unknown, key: string) => Promise; + }; + } + )._adapter.put( + STORE_NAMES.VECTOR_CLOCK, + { clock: { a: 5 }, lastUpdate: Date.now() }, + SINGLETON_KEY, + ); + + await service.mergeRemoteOpClocks([ + createTestOperation({ clientId: 'b', vectorClock: { b: 1 } }), + ]); + + const clock = await service.getVectorClock(); + expect(clock).toEqual({ a: 5, b: 1 }); + }); + }); + describe('mergeRemoteOpClocks', () => { it('should merge remote ops clocks into local clock', async () => { // Set initial local clock 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 8562353e0d..c1898e726b 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -45,6 +45,7 @@ import { IDB_OPEN_RETRIES_NON_LOCK, IDB_OPEN_RETRY_BASE_DELAY_MS, LOCK_NAMES, + MAX_VECTOR_CLOCK_SIZE, } from '../core/operation-log.const'; import { IndexedDBOpenError } from '../core/errors/indexed-db-open.error'; import { limitVectorClockSize, vectorClockToString } from '../../core/util/vector-clock'; @@ -890,6 +891,11 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); + // Pruned OUTSIDE the transaction (foreign awaits inside would break it); + // a stale author cannot be committed here — any interleaving append moves + // the op-log tail and the tail check below aborts the transaction (#9096). + const prunedVectorClock = await this.pruneClockForStorage(opts.vectorClock); + const storeNames: OpLogStoreName[] = [ STORE_NAMES.OPS, STORE_NAMES.STATE_CACHE, @@ -974,12 +980,12 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); + // The cached clock is restored as the DURABLE clock at hydration — prune + // with the same preserve set as every other durable write (#9096). + const vectorClock = await this.pruneClockForStorage(snapshot.vectorClock); await this._adapter.put(STORE_NAMES.STATE_CACHE, { id: SINGLETON_KEY, ...snapshot, + vectorClock, }); } @@ -2536,19 +2546,51 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + if (Object.keys(clock).length <= MAX_VECTOR_CLOCK_SIZE) { + return clock; + } + const currentClientId = await this.clientIdProvider.loadClientId(); + if (!currentClientId) { + return clock; + } + const importAuthorId = (await this.getLatestFullStateOp())?.clientId; + return limitVectorClockSize( + clock, + importAuthorId ? [currentClientId, importAuthorId] : [currentClientId], + ); + } + /** * Sets the vector clock directly. Used for: * - Migration from pf.META_MODEL on upgrade * - Sync import when receiving full state + * - Restoring the snapshot clock at hydration + * + * Prunes internally via {@link pruneClockForStorage} (#9096). */ async setVectorClock(clock: VectorClock): Promise { await this._ensureInit(); + const clockToStore = await this.pruneClockForStorage(clock); await this._adapter.put( STORE_NAMES.VECTOR_CLOCK, - { clock, lastUpdate: Date.now() }, + { clock: clockToStore, lastUpdate: Date.now() }, SINGLETON_KEY, ); - this._vectorClockCache = clock; + this._vectorClockCache = clockToStore; } /** @@ -2589,6 +2631,10 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { @@ -2596,16 +2642,6 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + const currentEntry = await tx.get( + STORE_NAMES.VECTOR_CLOCK, + SINGLETON_KEY, + ); + clockBefore = currentEntry?.clock ?? {}; + const storedImportAuthorId = await this._getLatestFullStateAuthorInTx(tx); + clockToStore = calculateRemoteClockMerge(clockBefore, ops, { + currentClientId, + storedImportAuthorId, + }); + await tx.put( + STORE_NAMES.VECTOR_CLOCK, + { clock: clockToStore, lastUpdate: Date.now() } satisfies VectorClockEntry, + SINGLETON_KEY, + ); + }, ); this._vectorClockCache = clockToStore; + + if (fullStateOp) { + Log.log( + `[OpLogStore] mergeRemoteOpClocks: REPLACED clock for FULL-STATE op ${fullStateOp.opType}\n` + + ` Op ID: ${fullStateOp.id}\n` + + ` Op clientId: ${fullStateOp.clientId}\n` + + ` Old clock (${Object.keys(clockBefore).length} entries): ${vectorClockToString(clockBefore)}\n` + + ` New clock (${Object.keys(clockToStore).length} entries): ${vectorClockToString(clockToStore)}`, + ); + } + Log.debug( + `[OpLogStore] mergeRemoteOpClocks: merged ${ops.length} remote ops\n` + + ` Clock before: ${vectorClockToString(clockBefore)}\n` + + ` Clock after: ${vectorClockToString(clockToStore)}`, + ); } /** diff --git a/src/app/op-log/persistence/sync-hydration.service.spec.ts b/src/app/op-log/persistence/sync-hydration.service.spec.ts index d7952dd8bd..cf4251a2f0 100644 --- a/src/app/op-log/persistence/sync-hydration.service.spec.ts +++ b/src/app/op-log/persistence/sync-hydration.service.spec.ts @@ -8,19 +8,14 @@ import { ClientIdService } from '../../core/util/client-id.service'; import { VectorClockService } from '../sync/vector-clock.service'; import { ValidateStateService } from '../validation/validate-state.service'; import { loadAllData } from '../../root-store/meta/load-all-data.action'; -import { - ActionType, - Operation, - OperationLogEntry, - OpType, -} from '../core/operation.types'; +import { ActionType, OperationLogEntry, OpType } from '../core/operation.types'; import { SyncProviderId } from '../sync-providers/provider.const'; import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const'; import { LOCAL_ONLY_SYNC_KEYS } from '../../features/config/local-only-sync-settings.util'; import { SnackService } from '../../core/snack/snack.service'; import { ArchiveDbAdapter } from '../../core/persistence/archive-db-adapter.service'; import { LockService } from '../sync/lock.service'; -import { LOCK_NAMES, MAX_VECTOR_CLOCK_SIZE } from '../core/operation-log.const'; +import { LOCK_NAMES } from '../core/operation-log.const'; import { TaskTimeSyncService } from '../../features/tasks/task-time-sync.service'; describe('SyncHydrationService', () => { @@ -58,12 +53,13 @@ describe('SyncHydrationService', () => { 'loadStateCache', 'getUnsynced', 'markRejected', - 'getLatestFullStateOp', + 'pruneClockForStorage', ]); // Default: no unsynced ops (for tests that don't care about this) mockOpLogStore.getUnsynced.and.resolveTo([]); mockOpLogStore.markRejected.and.resolveTo(); - mockOpLogStore.getLatestFullStateOp.and.resolveTo(undefined); + // Store-owned pruning (#9096): pass-through by default. + mockOpLogStore.pruneClockForStorage.and.callFake(async (clock) => clock); mockStateSnapshotService = jasmine.createSpyObj('StateSnapshotService', [ 'getAllSyncModelDataFromStoreAsync', ]); @@ -629,26 +625,22 @@ describe('SyncHydrationService', () => { expect(mockOpLogStore.setVectorClock).not.toHaveBeenCalled(); }); - it('should keep the stored import author when pruning the bootstrap baseline clock (#9096)', async () => { + it('should route the bootstrap baseline clock through store-owned pruning (#9096)', async () => { // No SYNC_IMPORT is created on this branch, so a previously stored - // full-state op stays the filter baseline — its author must survive - // pruning of the durable clock committed here. - const bloatedClock: Record = { importAuthor: 1, localClient: 5 }; - for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 5; i++) { - bloatedClock[`client-${i}`] = i + 100; - } - mockVectorClockService.getCurrentVectorClock.and.resolveTo(bloatedClock); - mockOpLogStore.getLatestFullStateOp.and.resolveTo({ - clientId: 'importAuthor', - } as Operation); + // full-state op stays the filter baseline — the durable clock + // committed here must go through pruneClockForStorage (which + // preserves the import author; behavior covered in the store spec). + mockVectorClockService.getCurrentVectorClock.and.resolveTo({ localClient: 5 }); + const prunedSentinel = { localClient: 6, importAuthor: 1 }; + mockOpLogStore.pruneClockForStorage.and.resolveTo(prunedSentinel); await service.hydrateFromRemoteSync({ task: {} }, undefined, false); + const pruneArg = mockOpLogStore.pruneClockForStorage.calls.mostRecent().args[0]; + expect(pruneArg['localClient']).toBe(6); // incremented BEFORE pruning const commitArg = mockOpLogStore.commitFileSnapshotBaseline.calls.mostRecent().args[0]; - expect(Object.keys(commitArg.vectorClock).length).toBe(MAX_VECTOR_CLOCK_SIZE); - expect(commitArg.vectorClock['importAuthor']).toBe(1); - expect(commitArg.vectorClock['localClient']).toBe(6); + expect(commitArg.vectorClock).toBe(prunedSentinel); }); it('should still dispatch loadAllData when createSyncImportOp is false', async () => { diff --git a/src/app/op-log/persistence/sync-hydration.service.ts b/src/app/op-log/persistence/sync-hydration.service.ts index 285d36abbf..ebdeb8ea81 100644 --- a/src/app/op-log/persistence/sync-hydration.service.ts +++ b/src/app/op-log/persistence/sync-hydration.service.ts @@ -11,11 +11,7 @@ import { SyncSessionValidationService } from '../sync/sync-session-validation.se import { loadAllData } from '../../root-store/meta/load-all-data.action'; import { Operation, OpType, ActionType, SyncImportReason } from '../core/operation.types'; import { uuidv7 } from '../../util/uuid-v7'; -import { - incrementVectorClock, - limitVectorClockSize, - mergeVectorClocks, -} from '../../core/util/vector-clock'; +import { incrementVectorClock, mergeVectorClocks } from '../../core/util/vector-clock'; import { OpLog } from '../../core/log'; import { AppDataComplete } from '../model/model-config'; import { selectSyncConfig } from '../../features/config/store/global-config.reducer'; @@ -204,16 +200,13 @@ export class SyncHydrationService { }); } - // On the file-snapshot bootstrap branch (no SYNC_IMPORT created below) - // this clock is committed as the DURABLE clock while a previously stored - // full-state op stays the filter baseline — its author must survive - // pruning (#9096). On the import branch self becomes the new author and - // is preserved either way. - const storedImportAuthorId = (await this.opLogStore.getLatestFullStateOp()) - ?.clientId; - const newClock = limitVectorClockSize( + // Store-owned pruning (#9096): preserves the current client and — for + // the file-snapshot bootstrap branch, where no SYNC_IMPORT is created + // and a previously stored full-state op stays the filter baseline — the + // latest full-state author. Must run after getOrGenerateClientId() so + // the id is persisted for the store's lookup. + const newClock = await this.opLogStore.pruneClockForStorage( incrementVectorClock(mergedClock, clientId), - storedImportAuthorId ? [clientId, storedImportAuthorId] : [clientId], ); let lastSeq: number; diff --git a/src/app/op-log/sync/server-migration.service.spec.ts b/src/app/op-log/sync/server-migration.service.spec.ts index 78e35f2bf3..57fc6c6e11 100644 --- a/src/app/op-log/sync/server-migration.service.spec.ts +++ b/src/app/op-log/sync/server-migration.service.spec.ts @@ -94,7 +94,10 @@ describe('ServerMigrationService', () => { 'hasSyncedOps', 'append', 'getOpsAfterSeq', + 'pruneClockForStorage', ]); + // Store-owned pruning (#9096): pass-through by default. + opLogStoreSpy.pruneClockForStorage.and.callFake(async (clock) => clock); vectorClockServiceSpy = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', ]); diff --git a/src/app/op-log/sync/server-migration.service.ts b/src/app/op-log/sync/server-migration.service.ts index caf406d4fa..3469529431 100644 --- a/src/app/op-log/sync/server-migration.service.ts +++ b/src/app/op-log/sync/server-migration.service.ts @@ -6,11 +6,7 @@ import { firstValueFrom } from 'rxjs'; import { OperationSyncCapable } from '../sync-providers/provider.interface'; import { OperationLogStoreService } from '../persistence/operation-log-store.service'; import { VectorClockService } from './vector-clock.service'; -import { - incrementVectorClock, - limitVectorClockSize, - mergeVectorClocks, -} from '../../core/util/vector-clock'; +import { incrementVectorClock, mergeVectorClocks } from '../../core/util/vector-clock'; import { StateSnapshotService } from '../backup/state-snapshot.service'; import { ValidateStateService } from '../validation/validate-state.service'; import { SnackService } from '../../core/snack/snack.service'; @@ -297,11 +293,13 @@ export class ServerMigrationService { for (const entry of allLocalOps) { mergedClock = mergeVectorClocks(mergedClock, entry.op.vectorClock); } - // Self is the SYNC_IMPORT author here, so preserving it keeps the entry - // the sync-import filter's rescue predicate reads on peers. - const newClock = limitVectorClockSize(incrementVectorClock(mergedClock, clientId), [ - clientId, - ]); + // Store-owned pruning (#9096) preserves self — the author of the + // SYNC_IMPORT built here, whose entry the sync-import filter's rescue + // predicate reads on peers — and, harmlessly, the author of the stored + // import this one supersedes. + const newClock = await this.opLogStore.pruneClockForStorage( + incrementVectorClock(mergedClock, clientId), + ); OpLog.normal( `ServerMigrationService: Merged ${allLocalOps.length} local op clocks into SYNC_IMPORT vector clock.`, diff --git a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts index 45ce830771..3bd4870a1c 100644 --- a/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts +++ b/src/app/op-log/testing/integration/post-sync-validation.integration.spec.ts @@ -189,12 +189,15 @@ describe('Post-sync validation latch (#7330) — integration', () => { 'append', 'loadStateCache', 'commitFileSnapshotBaseline', - 'getLatestFullStateOp', + 'pruneClockForStorage', ]); opLogStoreHydrationSpy.getLastSeq.and.resolveTo(0); opLogStoreHydrationSpy.getUnsynced.and.resolveTo([]); opLogStoreHydrationSpy.loadStateCache.and.resolveTo(null); - opLogStoreHydrationSpy.getLatestFullStateOp.and.resolveTo(undefined); + // Store-owned pruning (#9096): pass-through by default. + opLogStoreHydrationSpy.pruneClockForStorage.and.callFake( + async (clock: Record) => clock, + ); opLogStoreHydrationSpy.commitFileSnapshotBaseline.and.resolveTo({ seqs: [], writtenOps: [],