mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
refactor(sync): centralize clock pruning in store, make merge atomic (#9107)
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.
This commit is contained in:
Johannes Millan
GitHub
parent
fcbb789746
commit
fdbc2c1a86
15 changed files with 312 additions and 327 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<string, number> = {};
|
||||
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<string, number> = { 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();
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, number> = {};
|
||||
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<string, number> = { 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', () => {
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, number> = {};
|
||||
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();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<unknown>;
|
||||
};
|
||||
}
|
||||
)._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
|
||||
|
|
|
|||
|
|
@ -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<O
|
|||
}): Promise<{ seqs: number[]; writtenOps: Operation[]; skippedCount: number }> {
|
||||
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<O
|
|||
id: SINGLETON_KEY,
|
||||
state: opts.state,
|
||||
lastAppliedOpSeq: snapshotFrontier,
|
||||
vectorClock: opts.vectorClock,
|
||||
vectorClock: prunedVectorClock,
|
||||
compactedAt: opts.compactedAt,
|
||||
} satisfies StateCacheEntry);
|
||||
await tx.put(
|
||||
STORE_NAMES.VECTOR_CLOCK,
|
||||
{ clock: opts.vectorClock, lastUpdate: opts.compactedAt },
|
||||
{ clock: prunedVectorClock, lastUpdate: opts.compactedAt },
|
||||
SINGLETON_KEY,
|
||||
);
|
||||
if (opts.archiveYoung !== undefined) {
|
||||
|
|
@ -1001,7 +1007,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
}),
|
||||
);
|
||||
|
||||
this._vectorClockCache = { ...opts.vectorClock };
|
||||
this._vectorClockCache = { ...prunedVectorClock };
|
||||
this._invalidateAppliedAndUnsyncedCaches();
|
||||
return result;
|
||||
} catch (e) {
|
||||
|
|
@ -1948,9 +1954,13 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
snapshotEntityKeys?: string[];
|
||||
}): Promise<void> {
|
||||
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<O
|
|||
return this._vectorClockCache ? { ...this._vectorClockCache } : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Prunes a vector clock for durable storage — the single choke point for
|
||||
* client-side pruning (#9096). Preserves the current client and the latest
|
||||
* full-state author: the author's counter is low after the post-import
|
||||
* reset, so uploader-only pruning would evict exactly the entry the
|
||||
* sync-import filter's rescue predicate reads, and the server never
|
||||
* re-invents absent entries.
|
||||
*
|
||||
* Every durable-clock write in this store routes through this method, so
|
||||
* callers never prune; importing `limitVectorClockSize` outside this service
|
||||
* is lint-restricted. No-op for clocks within MAX_VECTOR_CLOCK_SIZE and when
|
||||
* no client id is available.
|
||||
*/
|
||||
async pruneClockForStorage(clock: VectorClock): Promise<VectorClock> {
|
||||
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<void> {
|
||||
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<O
|
|||
* Full-state ops reset the clock at their position in the batch. Operations
|
||||
* after the final reset are merged onto that new epoch in order.
|
||||
*
|
||||
* Runs as ONE readwrite transaction with a fresh in-transaction read of the
|
||||
* durable clock (never the per-tab cache) — a read-compute-put across
|
||||
* separate transactions loses entries when another tab writes in between.
|
||||
*
|
||||
* @param ops Remote operations whose clocks should be merged into local clock
|
||||
*/
|
||||
async mergeRemoteOpClocks(ops: Operation[]): Promise<void> {
|
||||
|
|
@ -2596,16 +2642,6 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
|
||||
await this._ensureInit();
|
||||
|
||||
// Get current local clock
|
||||
const currentClock = (await this.getVectorClock()) ?? {};
|
||||
|
||||
// DIAGNOSTIC LOGGING: Log current clock before merge
|
||||
Log.debug(
|
||||
`[OpLogStore] mergeRemoteOpClocks: BEFORE merge\n` +
|
||||
` Current clock: ${vectorClockToString(currentClock)}\n` +
|
||||
` Merging ${ops.length} remote ops`,
|
||||
);
|
||||
|
||||
let fullStateOp: Operation | undefined;
|
||||
for (const op of ops) {
|
||||
if (FULL_STATE_OP_TYPES.has(op.opType)) {
|
||||
|
|
@ -2613,16 +2649,8 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
}
|
||||
}
|
||||
|
||||
if (fullStateOp) {
|
||||
Log.log(
|
||||
`[OpLogStore] mergeRemoteOpClocks: REPLACING clock for FULL-STATE op ${fullStateOp.opType}\n` +
|
||||
` Op ID: ${fullStateOp.id}\n` +
|
||||
` Op clientId: ${fullStateOp.clientId}\n` +
|
||||
` Old clock (${Object.keys(currentClock).length} entries): ${vectorClockToString(currentClock)}\n` +
|
||||
` New base clock: ${vectorClockToString(fullStateOp.vectorClock)}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Foreign awaits must stay OUTSIDE the transaction (IDB auto-commits,
|
||||
// SQLite would deadlock on the connection queue).
|
||||
const currentClientId = await this.clientIdProvider.loadClientId();
|
||||
if (!currentClientId) {
|
||||
Log.warn(
|
||||
|
|
@ -2631,32 +2659,45 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
);
|
||||
}
|
||||
|
||||
const storedImportAuthorId = (await this.getLatestFullStateOp())?.clientId;
|
||||
const clockToStore = calculateRemoteClockMerge(currentClock, ops, {
|
||||
currentClientId,
|
||||
storedImportAuthorId,
|
||||
});
|
||||
|
||||
if (fullStateOp && currentClientId) {
|
||||
Log.log(
|
||||
`[OpLogStore] mergeRemoteOpClocks: RESET clock to minimal after ${fullStateOp.opType}\n` +
|
||||
` Minimal clock (${Object.keys(clockToStore).length} entries): ${vectorClockToString(clockToStore)}`,
|
||||
);
|
||||
}
|
||||
|
||||
// DIAGNOSTIC LOGGING: Log merged clock after merge
|
||||
Log.debug(
|
||||
`[OpLogStore] mergeRemoteOpClocks: AFTER merge\n` +
|
||||
` Merged clock: ${vectorClockToString(clockToStore)}`,
|
||||
);
|
||||
|
||||
// Update the vector clock store
|
||||
await this._adapter.put(
|
||||
STORE_NAMES.VECTOR_CLOCK,
|
||||
{ clock: clockToStore, lastUpdate: Date.now() },
|
||||
SINGLETON_KEY,
|
||||
let clockBefore: VectorClock = {};
|
||||
let clockToStore: VectorClock = {};
|
||||
await this._adapter.transaction(
|
||||
[STORE_NAMES.OPS, STORE_NAMES.VECTOR_CLOCK, STORE_NAMES.META],
|
||||
'readwrite',
|
||||
async (tx) => {
|
||||
const currentEntry = await tx.get<VectorClockEntry>(
|
||||
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)}`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<string, number> = { 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 () => {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
]);
|
||||
|
|
|
|||
|
|
@ -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.`,
|
||||
|
|
|
|||
|
|
@ -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<string, number>) => clock,
|
||||
);
|
||||
opLogStoreHydrationSpy.commitFileSnapshotBaseline.and.resolveTo({
|
||||
seqs: [],
|
||||
writtenOps: [],
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue