diff --git a/docs/sync-and-op-log/vector-clocks.md b/docs/sync-and-op-log/vector-clocks.md index dc961f2d64..bab29beee7 100644 --- a/docs/sync-and-op-log/vector-clocks.md +++ b/docs/sync-and-op-log/vector-clocks.md @@ -142,26 +142,27 @@ 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 passes `[currentClientId]` as the preserve list. +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). ### 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** `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 | +| 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 | ### Pruning is Rare -With MAX=20, a user needs 21+ unique client IDs before pruning triggers. The server preserves both the uploader and the latest causal full-state author when they are present in the incoming clock. Preserving that boundary edge prevents high-counter historical clients from making a post-import operation appear concurrent to the importing client. Other pruned edges can still cause one extra server round-trip (false CONCURRENT → client resolves → re-uploads with >MAX clock → GREATER_THAN → accepted). +With MAX=20, a user needs 21+ unique client IDs before pruning triggers. Both sides preserve the latest causal full-state author alongside their own id: the server when storing uploaded ops, the client at every site that prunes the durable clock (#9096). Preserving that boundary edge matters because `classifyOpAgainstSyncImport` rescues a post-import op from a different client via exactly one predicate — `op.vectorClock[importAuthor] >= importCounter` — and `limitVectorClockSize` never re-invents an absent entry, so an author dropped from the client's durable clock would be missing from every subsequent op permanently. Other pruned edges can still cause one extra server round-trip (false CONCURRENT → client resolves → re-uploads with >MAX clock → GREATER_THAN → accepted). --- @@ -221,7 +222,7 @@ An import is an explicit user action to restore **all clients** to a specific st | `BACKUP_IMPORT` (clean slate) | `BackupService` | Fresh clock `{newClientId: 1}` — small, no pruning issues | | Server migration | `ServerMigrationService` | Merge all local op clocks + global clock → increment → prune to MAX | | Sync hydration (conflict resolution) | `SyncHydrationService` | Merge local clock + state cache clock + remote snapshot clock → increment → prune to MAX | -| Auto-repair | `RepairOperationService` | Get current global clock → increment → prune to MAX | +| Auto-repair | `RepairOperationService` | Get current global clock → increment; ships the full clock unpruned (server prunes) | ### Full-State Operations Skip Server Conflict Detection diff --git a/packages/sync-core/src/vector-clock.ts b/packages/sync-core/src/vector-clock.ts index 909adf95f7..3cef2919b3 100644 --- a/packages/sync-core/src/vector-clock.ts +++ b/packages/sync-core/src/vector-clock.ts @@ -55,8 +55,11 @@ export const MAX_VECTOR_CLOCK_SIZE = 20; * ensure consistency. * * Standard vector clock comparison: missing keys mean "genuinely zero". - * With MAX_VECTOR_CLOCK_SIZE=20, pruning is extremely rare (needs 21+ unique - * client IDs), so pruning-aware comparison is unnecessary. + * Comparison stays pruning-unaware by design. Pruning is rare (needs 21+ + * unique client IDs) but real; when it fires, correctness rests on the + * preserve set — both server and client keep their own id plus the active + * full-state author (#9089, #9096) — and on the counter-based rescue + * predicates in sync-import-filter.ts absorbing residual pruning artifacts. * * @param a First vector clock * @param b Second vector clock diff --git a/src/app/core/util/vector-clock.spec.ts b/src/app/core/util/vector-clock.spec.ts index ba61eb4843..4710283c0b 100644 --- a/src/app/core/util/vector-clock.spec.ts +++ b/src/app/core/util/vector-clock.spec.ts @@ -12,7 +12,7 @@ describe('vector-clock', () => { describe('limitVectorClockSize', () => { it('should return clock unchanged when size is within limit', () => { const clock = { clientA: 5, clientB: 3, clientC: 2 }; - const result = limitVectorClockSize(clock, 'clientA'); + const result = limitVectorClockSize(clock, ['clientA']); expect(result).toEqual(clock); }); @@ -24,7 +24,7 @@ describe('vector-clock', () => { clock[`client_${i}`] = 100 + i; // All have higher counters } - const result = limitVectorClockSize(clock, 'currentClient'); + const result = limitVectorClockSize(clock, ['currentClient']); expect(Object.keys(result).length).toBeLessThanOrEqual(MAX_VECTOR_CLOCK_SIZE); expect(result['currentClient']).toBe(1); @@ -44,7 +44,7 @@ describe('vector-clock', () => { } clock[currentClientId] = 9747; // Current client has high counter - const result = limitVectorClockSize(clock, currentClientId); + const result = limitVectorClockSize(clock, [currentClientId]); expect(Object.keys(result).length).toBeLessThanOrEqual(MAX_VECTOR_CLOCK_SIZE); expect(result[currentClientId]).toBe(9747); @@ -52,6 +52,21 @@ describe('vector-clock', () => { expect(result['clientA']).toBeUndefined(); }); + it('should preserve every id in the preserve list, not just the first (#9096)', () => { + // Import-author scenario: the author's counter is the lowest in the + // clock, so it survives only through the preserve list. + const clock: Record = { importAuthor: 1, currentClient: 2 }; + for (let i = 0; i < MAX_VECTOR_CLOCK_SIZE + 5; i++) { + clock[`client_${i}`] = 100 + i; + } + + const result = limitVectorClockSize(clock, ['currentClient', 'importAuthor']); + + expect(Object.keys(result).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(result['currentClient']).toBe(2); + expect(result['importAuthor']).toBe(1); + }); + it('should limit to MAX_VECTOR_CLOCK_SIZE entries', () => { const currentClientId = 'current'; @@ -64,7 +79,7 @@ describe('vector-clock', () => { clock[`client_${i}`] = 100 + i; } - const result = limitVectorClockSize(clock, currentClientId); + const result = limitVectorClockSize(clock, [currentClientId]); expect(Object.keys(result).length).toBeLessThanOrEqual(MAX_VECTOR_CLOCK_SIZE); }); @@ -78,7 +93,7 @@ describe('vector-clock', () => { for (let i = 0; i < overflow; i++) { clock[`client_${i}`] = 100 + i; } - limitVectorClockSize(clock, 'current'); + limitVectorClockSize(clock, ['current']); sub.unsubscribe(); expect(events.length).toBe(1); @@ -89,7 +104,7 @@ describe('vector-clock', () => { it('should NOT emit on vectorClockPruned$ when within the limit', () => { const events: unknown[] = []; const sub = vectorClockPruned$.subscribe((e) => events.push(e)); - limitVectorClockSize({ a: 1, b: 2 }, 'a'); + limitVectorClockSize({ a: 1, b: 2 }, ['a']); sub.unsubscribe(); expect(events.length).toBe(0); }); @@ -105,7 +120,7 @@ describe('vector-clock', () => { clock[`client_${i}`] = 100 + i; } - limitVectorClockSize(clock, 'current'); + limitVectorClockSize(clock, ['current']); expect(warnSpy).toHaveBeenCalledTimes(1); const payload = warnSpy.calls.mostRecent().args[1] as { diff --git a/src/app/core/util/vector-clock.ts b/src/app/core/util/vector-clock.ts index 1704b3b754..2d2fe2d764 100644 --- a/src/app/core/util/vector-clock.ts +++ b/src/app/core/util/vector-clock.ts @@ -295,19 +295,21 @@ export const vectorClockPruned$ = new Subject<{ * Wraps the shared implementation from @sp/sync-core with client-side logging. * * @param clock The vector clock to limit - * @param currentClientId The current client's ID (always preserved) + * @param preserveClientIds Client IDs to always keep when present — the current + * client, plus the latest full-state import author so post-import ops keep + * proving causal knowledge of the import (#9096) * @returns A vector clock with at most MAX_VECTOR_CLOCK_SIZE entries */ export const limitVectorClockSize = ( clock: VectorClock, - currentClientId: string, + preserveClientIds: string[], ): VectorClock => { const entries = Object.entries(clock); if (entries.length <= MAX_VECTOR_CLOCK_SIZE) { return clock; } - const limited = sharedLimitVectorClockSize(clock, [currentClientId]); + const limited = sharedLimitVectorClockSize(clock, preserveClientIds); const prunedIds = Object.keys(clock).filter((id) => !(id in limited)); // WARN (not info): pruning is meant to be rare, so when it fires it is worth @@ -318,7 +320,7 @@ export const limitVectorClockSize = ( OpLog.warn('Vector clock pruning triggered', { originalSize: entries.length, maxSize: MAX_VECTOR_CLOCK_SIZE, - currentClientId, + preserveClientIds, prunedCount: prunedIds.length, prunedIds, survivingIds: Object.keys(limited), 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 ac4cd284e5..7c6f7ac1b9 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 @@ -10,7 +10,7 @@ import { SLOW_COMPACTION_THRESHOLD_MS, } from '../core/operation-log.const'; import { CURRENT_SCHEMA_VERSION } from './schema-migration.service'; -import { OperationLogEntry, OpType } from '../core/operation.types'; +import { Operation, 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,7 +57,9 @@ describe('OperationLogCompactionService', () => { 'resetCompactionCounter', 'deleteOpsWhere', 'getPendingRemoteOps', + 'getLatestFullStateOp', ]); + mockOpLogStore.getLatestFullStateOp.and.resolveTo(undefined); mockLockService = jasmine.createSpyObj('LockService', ['request']); mockStateSnapshot = jasmine.createSpyObj('StateSnapshotService', [ 'getStateSnapshot', @@ -211,6 +213,27 @@ describe('OperationLogCompactionService', () => { 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 () => { const smallClock = { clientA: 10, clientB: 5 }; mockVectorClockService.getCurrentVectorClock.and.resolveTo(smallClock); 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 36dd1bffc9..40b76bbae5 100644 --- a/src/app/op-log/persistence/operation-log-compaction.service.ts +++ b/src/app/op-log/persistence/operation-log-compaction.service.ts @@ -149,10 +149,17 @@ export class OperationLogCompactionService { 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. + // 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, clientId) + ? limitVectorClockSize( + currentVectorClock, + importAuthorId ? [clientId, importAuthorId] : [clientId], + ) : currentVectorClock; // 3. Get lastSeq IMMEDIATELY before writing cache to minimize race window 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 3ca07602ce..47cf8a7c8c 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 @@ -358,6 +358,29 @@ describe('OperationLogHydratorService', () => { 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); + }); + it('should not prune vector clock when within MAX_VECTOR_CLOCK_SIZE', async () => { const smallClock = { clientA: 5, clientB: 3 }; const snapshot = createMockSnapshot({ vectorClock: smallClock }); 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 65bd1c6805..8355bbf8c6 100644 --- a/src/app/op-log/persistence/operation-log-hydrator.service.ts +++ b/src/app/op-log/persistence/operation-log-hydrator.service.ts @@ -207,9 +207,16 @@ export class OperationLogHydratorService { 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, clientId) + ? limitVectorClockSize( + snapshot.vectorClock, + importAuthorId ? [clientId, importAuthorId] : [clientId], + ) : snapshot.vectorClock; await this.opLogStore.setVectorClock(clockToRestore); OpLog.normal( 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 e9bc79bc85..85002c173b 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,7 +55,9 @@ describe('OperationLogSnapshotService', () => { 'clearStateCacheBackup', 'restoreStateCacheFromBackup', 'getLastSeq', + 'getLatestFullStateOp', ]); + mockOpLogStore.getLatestFullStateOp.and.resolveTo(undefined); mockVectorClockService = jasmine.createSpyObj('VectorClockService', [ 'getCurrentVectorClock', ]); 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 13770744d0..aea88065f7 100644 --- a/src/app/op-log/persistence/operation-log-snapshot.service.ts +++ b/src/app/op-log/persistence/operation-log-snapshot.service.ts @@ -154,10 +154,17 @@ export class OperationLogSnapshotService { // 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. + // 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, clientId) + ? limitVectorClockSize( + vectorClock, + importAuthorId ? [clientId, importAuthorId] : [clientId], + ) : vectorClock; // Extract entity keys for conflict detection after compaction 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 95a263ad5b..8ad2865283 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 @@ -3965,6 +3965,96 @@ 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. + await service.append(createImportOp('importAuthor', 1), 'remote'); + await service.setVectorClock({ importAuthor: 1, testClient: 50 }); + + await service.mergeRemoteOpClocks(createBusyClientOps(MAX_VECTOR_CLOCK_SIZE)); + + const clock = await service.getVectorClock(); + expect(Object.keys(clock!).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock!['importAuthor']).toBe(1); + expect(clock!['testClient']).toBe(50); + }); + + it('should keep an in-batch import author when later ops in the same batch overflow the clock', async () => { + await service.setVectorClock({ testClient: 50 }); + + await service.mergeRemoteOpClocks([ + createImportOp('importAuthor', 1), + ...createBusyClientOps(MAX_VECTOR_CLOCK_SIZE + 1), + ]); + + const clock = await service.getVectorClock(); + expect(Object.keys(clock!).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock!['importAuthor']).toBe(1); + }); + + it('should keep the stored import author when the reducer checkpoint prunes the merged clock', async () => { + await service.append(createImportOp('importAuthor', 1), 'remote'); + await service.setVectorClock({ importAuthor: 1, testClient: 50 }); + + const busyOps = createBusyClientOps(MAX_VECTOR_CLOCK_SIZE); + const seqs: number[] = []; + for (const op of busyOps) { + seqs.push(await service.append(op, 'remote', { pendingApply: true })); + } + + await service.markReducersCommittedAndMergeClocks(seqs, busyOps); + + const clock = await service.getVectorClock(); + expect(Object.keys(clock!).length).toBe(MAX_VECTOR_CLOCK_SIZE); + expect(clock!['importAuthor']).toBe(1); + }); + + it('should protect the previous active import author when the latest import is rejected in the same checkpoint', async () => { + // The author must be resolved INSIDE the checkpoint transaction, after + // rejections are written — a pre-transaction read would still name the + // about-to-be-rejected import's author and let the real baseline's + // author be evicted. + await service.append(createImportOp('olderAuthor', 1), 'remote'); + const rejectedImport = createImportOp('rejectedAuthor', 2); + await service.append(rejectedImport, 'remote', { pendingApply: true }); + await service.setVectorClock({ olderAuthor: 1, rejectedAuthor: 2, testClient: 50 }); + + const busyOps = createBusyClientOps(MAX_VECTOR_CLOCK_SIZE); + const seqs: number[] = []; + for (const op of busyOps) { + seqs.push(await service.append(op, 'remote', { pendingApply: true })); + } + + await service.markReducersCommittedAndMergeClocks(seqs, busyOps, [ + rejectedImport.id, + ]); + + const clock = await service.getVectorClock(); + expect(clock!['olderAuthor']).toBe(1); + expect((await service.getLatestFullStateOpEntry())?.op.clientId).toBe( + 'olderAuthor', + ); + }); + }); + describe('mergeRemoteOpClocks', () => { it('should merge remote ops clocks into local clock', async () => { // Set initial local clock @@ -4147,7 +4237,7 @@ describe('OperationLogStoreService', () => { const newOpClock = incrementVectorClock(storedClock, 'localClient'); // Simulate server-side pruning (server prunes to MAX_VECTOR_CLOCK_SIZE) - const prunedClock = limitVectorClockSize(newOpClock, 'localClient'); + const prunedClock = limitVectorClockSize(newOpClock, ['localClient']); // importClient must survive pruning expect(prunedClock['importClient']).toBe(1); 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 880525356c..8562353e0d 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -181,16 +181,28 @@ const getStoredOpType = (op: Operation | CompactOperation): string => * * Kept pure so both the standalone merge path and the atomic reducer checkpoint * use exactly the same full-state reset and pruning semantics. + * + * Pruning preserves the latest full-state author alongside the current client + * (#9096): the import author's counter is low after the post-import reset, so + * uploader-only pruning would evict exactly the entry the sync-import filter's + * `knows-import-counter` rescue reads — and the server never re-invents absent + * entries. A full-state op inside `ops` supersedes `storedImportAuthorId`. */ const calculateRemoteClockMerge = ( currentClock: VectorClock, ops: readonly Operation[], - currentClientId: string | null, + opts: { + currentClientId: string | null; + storedImportAuthorId: string | undefined; + }, ): VectorClock => { + const { currentClientId } = opts; + let importAuthorId = opts.storedImportAuthorId; let mergedClock: VectorClock = { ...currentClock }; for (const op of ops) { if (FULL_STATE_OP_TYPES.has(op.opType)) { + importAuthorId = op.clientId; const clockBeforeReset = mergedClock; if (!currentClientId) { mergedClock = { ...op.vectorClock }; @@ -218,9 +230,14 @@ const calculateRemoteClockMerge = ( } } - return currentClientId - ? limitVectorClockSize(mergedClock, currentClientId) - : mergedClock; + // No client ID → no pruning at all (never prune with the author id alone). + if (!currentClientId) { + return mergedClock; + } + return limitVectorClockSize( + mergedClock, + importAuthorId ? [currentClientId, importAuthorId] : [currentClientId], + ); }; // Note: DBSchema requires literal string keys matching STORE_NAMES values @@ -655,6 +672,29 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + const meta = await this._getFullStateOpsMetaInTxOrRebuild(tx); + const refsNewestFirst = [...meta.refs].sort((a, b) => b.seq - a.seq); + for (const ref of refsNewestFirst) { + const stored = await tx.get(STORE_NAMES.OPS, ref.seq); + if ( + !stored || + stored.rejectedAt !== undefined || + getOpId(stored.op) !== ref.opId || + !isFullStateOpType(getStoredOpType(stored.op)) + ) { + continue; + } + return decodeStoredEntry(stored).op.clientId; + } + return undefined; + } + private async _rebuildFullStateOpsMeta(): Promise { const refs: FullStateOpRef[] = []; await this._adapter.iterate( @@ -1214,11 +1254,6 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort(STORE_NAMES.OPS, seq); @@ -1257,6 +1292,14 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { @@ -53,10 +58,12 @@ describe('SyncHydrationService', () => { 'loadStateCache', 'getUnsynced', 'markRejected', + 'getLatestFullStateOp', ]); // 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); mockStateSnapshotService = jasmine.createSpyObj('StateSnapshotService', [ 'getAllSyncModelDataFromStoreAsync', ]); @@ -622,6 +629,28 @@ describe('SyncHydrationService', () => { expect(mockOpLogStore.setVectorClock).not.toHaveBeenCalled(); }); + it('should keep the stored import author when pruning the bootstrap baseline clock (#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); + + await service.hydrateFromRemoteSync({ task: {} }, undefined, false); + + 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); + }); + it('should still dispatch loadAllData when createSyncImportOp is false', async () => { const downloadedData = { task: { ids: ['t1'] } }; diff --git a/src/app/op-log/persistence/sync-hydration.service.ts b/src/app/op-log/persistence/sync-hydration.service.ts index 8ac6dea6ef..285d36abbf 100644 --- a/src/app/op-log/persistence/sync-hydration.service.ts +++ b/src/app/op-log/persistence/sync-hydration.service.ts @@ -204,9 +204,16 @@ 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( incrementVectorClock(mergedClock, clientId), - clientId, + storedImportAuthorId ? [clientId, storedImportAuthorId] : [clientId], ); let lastSeq: number; diff --git a/src/app/op-log/sync/server-migration.service.ts b/src/app/op-log/sync/server-migration.service.ts index a2f3dd1320..caf406d4fa 100644 --- a/src/app/op-log/sync/server-migration.service.ts +++ b/src/app/op-log/sync/server-migration.service.ts @@ -297,10 +297,11 @@ export class ServerMigrationService { for (const entry of allLocalOps) { mergedClock = mergeVectorClocks(mergedClock, entry.op.vectorClock); } - const newClock = limitVectorClockSize( - incrementVectorClock(mergedClock, clientId), + // 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, - ); + ]); OpLog.normal( `ServerMigrationService: Merged ${allLocalOps.length} local op clocks into SYNC_IMPORT vector clock.`, diff --git a/src/app/op-log/testing/integration/import-sync.integration.spec.ts b/src/app/op-log/testing/integration/import-sync.integration.spec.ts index b4a5df1b8e..617698851f 100644 --- a/src/app/op-log/testing/integration/import-sync.integration.spec.ts +++ b/src/app/op-log/testing/integration/import-sync.integration.spec.ts @@ -543,7 +543,7 @@ describe('Import + Sync Integration', () => { }; // Simulate server-side pruning of B's op clock to MAX=20 - const prunedOpClock = limitVectorClockSize(postImportClock, clientBId); + const prunedOpClock = limitVectorClockSize(postImportClock, [clientBId]); // BUG: After pruning, import-fresh:1 gets dropped // (it has the lowest counter). Now compareVectorClocks returns CONCURRENT 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 196bac8ade..45ce830771 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,10 +189,12 @@ describe('Post-sync validation latch (#7330) — integration', () => { 'append', 'loadStateCache', 'commitFileSnapshotBaseline', + 'getLatestFullStateOp', ]); opLogStoreHydrationSpy.getLastSeq.and.resolveTo(0); opLogStoreHydrationSpy.getUnsynced.and.resolveTo([]); opLogStoreHydrationSpy.loadStateCache.and.resolveTo(null); + opLogStoreHydrationSpy.getLatestFullStateOp.and.resolveTo(undefined); opLogStoreHydrationSpy.commitFileSnapshotBaseline.and.resolveTo({ seqs: [], writtenOps: [], diff --git a/src/app/op-log/testing/integration/vector-clock-import-reset.integration.spec.ts b/src/app/op-log/testing/integration/vector-clock-import-reset.integration.spec.ts index adde32ba90..d411abf750 100644 --- a/src/app/op-log/testing/integration/vector-clock-import-reset.integration.spec.ts +++ b/src/app/op-log/testing/integration/vector-clock-import-reset.integration.spec.ts @@ -518,7 +518,7 @@ describe('Vector Clock Import Reset Integration', () => { // Server prunes using limitVectorClockSize, preserving the uploading // client's ID (this is what the real server does). - const prunedClock = limitVectorClockSize(newClientClock, newClientId); + const prunedClock = limitVectorClockSize(newClientClock, [newClientId]); // The pruned clock keeps the new client but drops the lowest import entry expect(Object.keys(prunedClock).length).toBe(MAX_VECTOR_CLOCK_SIZE);