From 807d3bf5e5aac0a265c31ff40f2a5504b07ed82f Mon Sep 17 00:00:00 2001 From: Symon Baikov Date: Wed, 17 Jun 2026 13:55:31 +0300 Subject: [PATCH] perf(sync): cache latest full-state op lookup (#8427) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * perf(sync): cache latest full-state op lookup * fix(sync): harden full-state op metadata * refactor(sync): dedupe full-state-ops meta helper + SQLite tests Extract the FullStateOpRef/FullStateOpsMetaEntry shape and the "latest = max UUIDv7" rule into full-state-ops-meta.ts, shared by the upgrade-time seed (db-upgrade.ts) and the runtime maintenance (operation-log-store.service.ts) so the two copies can't drift. The shared builder always copies refs, removing the prior store-vs-upgrade copy inconsistency (call sites already pass fresh arrays, so behavior is unchanged). Add store-service tests driving the full-state metadata pointer over a real SQLite engine (sql.js), including the rebuild-on-read fallback that keeps the pointer correct on SQLite, where the IndexedDB-only populate-on-upgrade seed never runs. * fix(sync): prune full-state meta atomically in clearFullStateOpsExcept Read the meta pointer INSIDE the delete transaction instead of taking a snapshot before it. The prior out-of-tx read could be clobbered by a full-state append committed between the read and the write, dropping that op's ref from the pointer while the op stayed in OPS — risking a stale filtering baseline on the sync-receive path (packages/sync-core caller). The OPERATION_LOG lock masks this today, but the method is now atomic and self-consistent regardless of lock scope, matching deleteOpsWhere. Also route the two hand-built meta writes (clearAllOperations, runDestructiveStateReplacement) through buildFullStateOpsMeta so `latest` is always derived from refs, never hand-asserted. --------- Co-authored-by: Johannes Millan --- src/app/op-log/persistence/db-keys.const.ts | 7 +- src/app/op-log/persistence/db-upgrade.spec.ts | 97 +++- src/app/op-log/persistence/db-upgrade.ts | 74 ++- .../op-log/persistence/full-state-ops-meta.ts | 48 ++ .../op-log/persistence/op-log-db-schema.ts | 4 +- .../operation-log-store.service.spec.ts | 415 +++++++++++++++- .../operation-log-store.service.ts | 443 +++++++++++++----- 7 files changed, 949 insertions(+), 139 deletions(-) create mode 100644 src/app/op-log/persistence/full-state-ops-meta.ts diff --git a/src/app/op-log/persistence/db-keys.const.ts b/src/app/op-log/persistence/db-keys.const.ts index fd190b8635..6d96c32a41 100644 --- a/src/app/op-log/persistence/db-keys.const.ts +++ b/src/app/op-log/persistence/db-keys.const.ts @@ -12,7 +12,7 @@ import { CompleteBackup } from '../sync-exports'; export const DB_NAME = 'SUP_OPS'; /** Current database schema version */ -export const DB_VERSION = 6; +export const DB_VERSION = 7; /** Object store names */ export const STORE_NAMES = { @@ -32,6 +32,8 @@ export const STORE_NAMES = { PROFILE_DATA: 'profile_data' as const, /** Client ID - sync device identity (singleton, key = SINGLETON_KEY) */ CLIENT_ID: 'client_id' as const, + /** Small persistence metadata records derived from ops */ + META: 'meta' as const, } as const; /** Common key used for singleton entries */ @@ -40,6 +42,9 @@ export const SINGLETON_KEY = 'current' as const; /** Backup key for state cache backup */ export const BACKUP_KEY = 'backup' as const; +/** Meta key for derived full-state operation refs */ +export const FULL_STATE_OPS_META_KEY = 'full_state_ops' as const; + /** Index names for ops object store */ export const OPS_INDEXES = { BY_ID: 'byId' as const, diff --git a/src/app/op-log/persistence/db-upgrade.spec.ts b/src/app/op-log/persistence/db-upgrade.spec.ts index 3f0710cb6e..cad46872c1 100644 --- a/src/app/op-log/persistence/db-upgrade.spec.ts +++ b/src/app/op-log/persistence/db-upgrade.spec.ts @@ -1,5 +1,7 @@ import { runDbUpgrade } from './db-upgrade'; -import { STORE_NAMES, OPS_INDEXES } from './db-keys.const'; +import { FULL_STATE_OPS_META_KEY, STORE_NAMES, OPS_INDEXES } from './db-keys.const'; +import { deleteDB, openDB } from 'idb'; +import { OpType } from '../core/operation.types'; describe('runDbUpgrade', () => { // Mock store with index tracking @@ -153,8 +155,9 @@ describe('runDbUpgrade', () => { // Version 2 upgrade doesn't create any stores (only version 3+ run) // Version 3 only adds an index, doesn't create stores - // Version 4 creates archive stores, version 5 profile_data, version 6 client_id - expect(db.createObjectStore).toHaveBeenCalledTimes(4); // archive_young, archive_old, profile_data, client_id + // Version 4 creates archive stores, version 5 profile_data, + // version 6 client_id, version 7 meta. + expect(db.createObjectStore).toHaveBeenCalledTimes(5); expect(db.createObjectStore).not.toHaveBeenCalledWith( STORE_NAMES.OPS, jasmine.anything(), @@ -210,17 +213,92 @@ describe('runDbUpgrade', () => { expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.CLIENT_ID); }); - it('should not recreate earlier stores', () => { + it('should not recreate stores before version 6', () => { const preExisting = new Map([[STORE_NAMES.OPS, { store: createMockStore() }]]); const { db, tx } = createMocks(preExisting); runDbUpgrade(db, 5, tx); - expect(db.createObjectStore).toHaveBeenCalledTimes(1); // client_id only + expect(db.createObjectStore).toHaveBeenCalledTimes(2); // client_id, meta + expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.CLIENT_ID); + expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.META); }); }); - describe('full upgrade path (version 0 to 6)', () => { + describe('version 7 upgrade (from version 6)', () => { + it('should create meta store', () => { + const preExisting = new Map([[STORE_NAMES.OPS, { store: createMockStore() }]]); + const { db, tx } = createMocks(preExisting); + + runDbUpgrade(db, 6, tx); + + expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.META); + }); + + it('should not recreate earlier stores', () => { + const preExisting = new Map([[STORE_NAMES.OPS, { store: createMockStore() }]]); + const { db, tx } = createMocks(preExisting); + + runDbUpgrade(db, 6, tx); + + expect(db.createObjectStore).toHaveBeenCalledTimes(1); + }); + + it('should populate full-state metadata from existing ops', async () => { + const dbName = `SUP_OPS_upgrade_meta_${Date.now()}_${Math.random()}`; + await deleteDB(dbName); + + const v6Db = await openDB(dbName, 6, { + upgrade: (db) => { + const opStore = db.createObjectStore(STORE_NAMES.OPS, { + keyPath: 'seq', + autoIncrement: true, + }); + opStore.createIndex(OPS_INDEXES.BY_ID, 'op.id', { unique: true }); + opStore.createIndex(OPS_INDEXES.BY_SYNCED_AT, 'syncedAt'); + opStore.createIndex(OPS_INDEXES.BY_SOURCE_AND_STATUS, [ + 'source', + 'applicationStatus', + ]); + }, + }); + await v6Db.add(STORE_NAMES.OPS, { + op: { id: '01900000-0000-7000-8000-000000000041', o: OpType.SyncImport }, + appliedAt: Date.now(), + source: 'local', + }); + await v6Db.add(STORE_NAMES.OPS, { + op: { id: '01900000-0000-7000-8000-000000000043', o: OpType.Update }, + appliedAt: Date.now(), + source: 'local', + }); + await v6Db.add(STORE_NAMES.OPS, { + op: { id: '01900000-0000-7000-8000-000000000042', o: OpType.BackupImport }, + appliedAt: Date.now(), + source: 'remote', + syncedAt: Date.now(), + }); + v6Db.close(); + + const v7Db = await openDB(dbName, 7, { + upgrade: (db, oldVersion, _newVersion, tx) => runDbUpgrade(db, oldVersion, tx), + }); + + const meta = await v7Db.get(STORE_NAMES.META, FULL_STATE_OPS_META_KEY); + expect(meta).toEqual({ + refs: [ + { opId: '01900000-0000-7000-8000-000000000041', seq: 1 }, + { opId: '01900000-0000-7000-8000-000000000042', seq: 3 }, + ], + latest: { opId: '01900000-0000-7000-8000-000000000042', seq: 3 }, + }); + + v7Db.close(); + await deleteDB(dbName); + }); + }); + + describe('full upgrade path (version 0 to 7)', () => { it('should create all stores and indexes when upgrading from version 0', () => { const { db, tx } = createMocks(); @@ -262,8 +340,11 @@ describe('runDbUpgrade', () => { // Version 6 store expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.CLIENT_ID); - // Total: 8 stores created - expect(db.createObjectStore).toHaveBeenCalledTimes(8); + // Version 7 store + expect(db.createObjectStore).toHaveBeenCalledWith(STORE_NAMES.META); + + // Total: 9 stores created + expect(db.createObjectStore).toHaveBeenCalledTimes(9); }); it('should create all indexes on ops store when upgrading from version 0', () => { diff --git a/src/app/op-log/persistence/db-upgrade.ts b/src/app/op-log/persistence/db-upgrade.ts index 67dd2e2488..cc532b280b 100644 --- a/src/app/op-log/persistence/db-upgrade.ts +++ b/src/app/op-log/persistence/db-upgrade.ts @@ -8,8 +8,71 @@ * This shared function ensures schema consistency regardless of service initialization order. */ -import { IDBPDatabase, IDBPTransaction } from 'idb'; -import { STORE_NAMES, OPS_INDEXES } from './db-keys.const'; +import { IDBPDatabase, IDBPTransaction, unwrap } from 'idb'; +import { FULL_STATE_OPS_META_KEY, STORE_NAMES, OPS_INDEXES } from './db-keys.const'; +import { isFullStateOpType } from '../core/operation.types'; +import { buildFullStateOpsMeta, FullStateOpRef } from './full-state-ops-meta'; + +const isRecord = (value: unknown): value is Record => + typeof value === 'object' && value !== null; + +const getStoredOpId = (op: unknown): string | undefined => { + if (!isRecord(op)) { + return undefined; + } + return typeof op['id'] === 'string' ? op['id'] : undefined; +}; + +const getStoredOpType = (op: unknown): string | undefined => { + if (!isRecord(op)) { + return undefined; + } + const compactType = op['o']; + if (typeof compactType === 'string') { + return compactType; + } + const fullType = op['opType']; + return typeof fullType === 'string' ? fullType : undefined; +}; + +const getFullStateRefFromStoredEntry = ( + storedEntry: unknown, + seq: number, +): FullStateOpRef | undefined => { + if (!isRecord(storedEntry)) { + return undefined; + } + const op = storedEntry['op']; + const opId = getStoredOpId(op); + const opType = getStoredOpType(op); + return opId && opType && isFullStateOpType(opType) ? { opId, seq } : undefined; +}; + +const populateFullStateOpsMetaDuringUpgrade = ( + transaction: IDBPTransaction, +): void => { + const opsStore = unwrap(transaction.objectStore(STORE_NAMES.OPS)) as IDBObjectStore; + const metaStore = unwrap(transaction.objectStore(STORE_NAMES.META)) as IDBObjectStore; + if (!opsStore?.openCursor || !metaStore?.put) { + return; + } + const refs: FullStateOpRef[] = []; + const request = opsStore.openCursor(); + + request.onsuccess = (): void => { + const cursor = request.result; + if (!cursor) { + metaStore.put(buildFullStateOpsMeta(refs), FULL_STATE_OPS_META_KEY); + return; + } + + const ref = getFullStateRefFromStoredEntry(cursor.value, Number(cursor.primaryKey)); + if (ref) { + refs.push(ref); + } + cursor.continue(); + }; +}; /** * Performs the database upgrade for SUP_OPS. @@ -76,4 +139,11 @@ export const runDbUpgrade = ( if (oldVersion < 6) { db.createObjectStore(STORE_NAMES.CLIENT_ID); } + + // Version 7: Add meta store for small derived pointers and seed it from + // existing ops before any post-upgrade write can observe an empty meta row. + if (oldVersion < 7) { + db.createObjectStore(STORE_NAMES.META); + populateFullStateOpsMetaDuringUpgrade(transaction); + } }; diff --git a/src/app/op-log/persistence/full-state-ops-meta.ts b/src/app/op-log/persistence/full-state-ops-meta.ts new file mode 100644 index 0000000000..9319e27474 --- /dev/null +++ b/src/app/op-log/persistence/full-state-ops-meta.ts @@ -0,0 +1,48 @@ +/** + * Shared shape + pure helpers for the derived "full-state op" metadata pointer + * stored at `SUP_OPS.meta` under `FULL_STATE_OPS_META_KEY`. + * + * Kept in one place so the `{ refs, latest }` shape and the "latest = max + * UUIDv7" rule cannot drift between the upgrade-time seed (`db-upgrade.ts`, + * which scans raw cursor values) and the runtime maintenance + * (`operation-log-store.service.ts`, which works with decoded ops). + */ + +/** A pointer to one full-state op: its `op.id` and its `seq` primary key. */ +export interface FullStateOpRef { + opId: string; + seq: number; +} + +/** + * The derived full-state metadata: every tracked full-state op ref plus the + * latest by UUIDv7. `latest` is always derived from `refs` (never trusted from + * storage), so a stale/corrupt stored `latest` self-corrects on the next build. + */ +export interface FullStateOpsMetaEntry { + refs: FullStateOpRef[]; + latest?: FullStateOpRef; +} + +/** + * The latest full-state op by UUIDv7. Lexicographic comparison is correct for + * UUIDv7 (time-ordered, lowercase hex), matching the pre-pointer full scan. + */ +export const getLatestFullStateRef = ( + refs: ReadonlyArray, +): FullStateOpRef | undefined => + refs.reduce( + (latest, ref) => (!latest || ref.opId > latest.opId ? ref : latest), + undefined, + ); + +/** + * Builds a meta entry from refs, deriving `latest`. Copies `refs` so a stored + * meta object can never alias a caller-owned array. + */ +export const buildFullStateOpsMeta = ( + refs: ReadonlyArray, +): FullStateOpsMetaEntry => { + const latest = getLatestFullStateRef(refs); + return latest ? { refs: [...refs], latest } : { refs: [...refs] }; +}; diff --git a/src/app/op-log/persistence/op-log-db-schema.ts b/src/app/op-log/persistence/op-log-db-schema.ts index 371c8b9580..3162278010 100644 --- a/src/app/op-log/persistence/op-log-db-schema.ts +++ b/src/app/op-log/persistence/op-log-db-schema.ts @@ -40,7 +40,7 @@ export interface OpLogDbSchema { } /** - * Current `SUP_OPS` schema, mirroring db-upgrade.ts (currently v6). + * Current `SUP_OPS` schema, mirroring db-upgrade.ts (currently v7). * * `name`/`version` are reused from `db-keys.const.ts` (not re-literaled) so the * adapter opens at exactly the version `runDbUpgrade` migrates to — a future @@ -78,5 +78,7 @@ export const OP_LOG_DB_SCHEMA: OpLogDbSchema = { { name: STORE_NAMES.PROFILE_DATA, keyPath: 'id' }, // keyless singleton: clientId stored under SINGLETON_KEY { name: STORE_NAMES.CLIENT_ID }, + // keyless singleton metadata records, written with explicit keys + { name: STORE_NAMES.META }, ], }; 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 235543934e..bb1bf1641d 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 @@ -20,13 +20,15 @@ import { limitVectorClockSize, MAX_VECTOR_CLOCK_SIZE } from '@sp/shared-schema'; import { CLIENT_ID_PROVIDER, ClientIdProvider } from '../util/client-id.provider'; import { OP_LOG_DB_ADAPTER_FACTORY } from './op-log-db-adapter.token'; import { OpLogDbAdapter } from './op-log-db-adapter'; +import { SqliteOpLogAdapter } from './sqlite-op-log-adapter'; +import { createSqlJsDb } from './sql-js-db.test-helper'; import { IDB_OPEN_RETRIES, IDB_OPEN_RETRIES_NON_LOCK, IDB_OPEN_RETRY_BASE_DELAY_MS, } from '../core/operation-log.const'; import { IndexedDBOpenError } from '../core/errors/indexed-db-open.error'; -import { SINGLETON_KEY, STORE_NAMES } from './db-keys.const'; +import { FULL_STATE_OPS_META_KEY, SINGLETON_KEY, STORE_NAMES } from './db-keys.const'; describe('OperationLogStoreService', () => { let service: OperationLogStoreService; @@ -501,6 +503,35 @@ describe('OperationLogStoreService', () => { expect(remaining.length).toBe(1); expect(remaining[0].op.actionType).toBe('[Task] Update' as ActionType); }); + + it('should clear full-state metadata when deleting a full-state op', async () => { + const syncImportOp = createTestOperation({ + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await service.append(syncImportOp); + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(syncImportOp.id); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + const transactionSpy = spyOn(adapter, 'transaction').and.callThrough(); + + await service.deleteOpsWhere((entry) => entry.op.id === syncImportOp.id); + + spyOn(adapter, 'iterate').and.callThrough(); + + expect(await service.getLatestFullStateOpEntry()).toBeUndefined(); + expect(transactionSpy).toHaveBeenCalledWith( + [STORE_NAMES.OPS, STORE_NAMES.META], + 'readwrite', + jasmine.any(Function), + ); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); }); describe('getLastSeq', () => { @@ -522,6 +553,292 @@ describe('OperationLogStoreService', () => { }); }); + describe('getLatestFullStateOpEntry', () => { + it('should read latest full-state op via metadata without scanning ops', async () => { + const oldImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000001', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + const regularOp = createTestOperation({ + id: '01900000-0000-7000-8000-000000000002', + }); + const latestImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000003', + opType: OpType.BackupImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + + await service.append(oldImport); + await service.append(regularOp); + await service.append(latestImport, 'remote'); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + const latestEntry = await service.getLatestFullStateOpEntry(); + + expect(latestEntry?.op.id).toBe(latestImport.id); + expect(latestEntry?.source).toBe('remote'); + expect(latestEntry?.syncedAt).toBeDefined(); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); + + it('should rebuild missing metadata once and use it for subsequent reads', async () => { + const latestImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000011', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await service.append( + createTestOperation({ id: '01900000-0000-7000-8000-000000000010' }), + ); + await service.append(latestImport); + + const db = ( + service as unknown as { + db: IDBPDatabase; + } + ).db; + await db.delete('meta', 'full_state_ops'); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(latestImport.id); + expect(adapter.iterate).toHaveBeenCalledTimes(1); + + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(latestImport.id); + expect(adapter.iterate).toHaveBeenCalledTimes(1); + }); + + it('should rebuild missing metadata inside a full-state append before recording the new ref', async () => { + const latestExistingImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000032', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + const lowerNewImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000031', + opType: OpType.BackupImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + + await service.append(latestExistingImport); + const db = ( + service as unknown as { + db: IDBPDatabase; + } + ).db; + await db.delete(STORE_NAMES.META, FULL_STATE_OPS_META_KEY); + + await service.append(lowerNewImport, 'remote'); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe( + latestExistingImport.id, + ); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); + + it('should rebuild malformed metadata instead of throwing', async () => { + const latestImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000041', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await service.append(latestImport); + + const db = ( + service as unknown as { + db: IDBPDatabase; + } + ).db; + await db.put(STORE_NAMES.META, { refs: 'not-an-array' }, FULL_STATE_OPS_META_KEY); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + await expectAsync(service.getLatestFullStateOpEntry()).toBeResolved(); + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(latestImport.id); + expect(adapter.iterate).toHaveBeenCalledTimes(1); + }); + + it('should delete stale full-state ops via metadata and keep the excluded op', async () => { + const staleImportA = createTestOperation({ + id: '01900000-0000-7000-8000-000000000021', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + const staleImportB = createTestOperation({ + id: '01900000-0000-7000-8000-000000000022', + opType: OpType.Repair, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + const keepImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000023', + opType: OpType.BackupImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + + await service.append(staleImportA); + await service.append( + createTestOperation({ id: '01900000-0000-7000-8000-000000000024' }), + ); + await service.append(staleImportB); + await service.append(keepImport); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + const deletedCount = await service.clearFullStateOpsExcept([keepImport.id]); + + expect(deletedCount).toBe(2); + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(keepImport.id); + expect((await service.getOpsAfterSeq(0)).map((entry) => entry.op.id)).toEqual([ + '01900000-0000-7000-8000-000000000024', + keepImport.id, + ]); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); + }); + + // The full-state metadata pointer is adapter-agnostic, but the rest of this + // suite drives it through the IndexedDB adapter. These tests pin the SAME + // behavior through the SQLite adapter (Android default, #8389) against a real + // engine (sql.js) — including the rebuild-on-read fallback, which is what + // keeps the pointer correct on SQLite (the IndexedDB-only populate-on-upgrade + // seed in db-upgrade.ts never runs there). + describe('full-state metadata over the SQLite backend', () => { + const freshSqliteService = async (): Promise<{ + svc: OperationLogStoreService; + adapter: OpLogDbAdapter; + }> => { + const adapter = new SqliteOpLogAdapter(await createSqlJsDb()); + TestBed.resetTestingModule(); + TestBed.configureTestingModule({ + providers: [ + OperationLogStoreService, + { provide: CLIENT_ID_PROVIDER, useValue: mockClientIdProvider }, + { provide: OP_LOG_DB_ADAPTER_FACTORY, useValue: () => adapter }, + ], + }); + const svc = TestBed.inject(OperationLogStoreService); + await svc.init(); + return { svc, adapter }; + }; + + it('tracks the latest full-state op by UUIDv7 without scanning', async () => { + const { svc, adapter } = await freshSqliteService(); + await svc.append( + createTestOperation({ + id: '01900000-0000-7000-8000-000000000001', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }), + ); + await svc.append( + createTestOperation({ id: '01900000-0000-7000-8000-000000000002' }), + ); + const latestImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000003', + opType: OpType.BackupImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await svc.append(latestImport, 'remote'); + + const iterateSpy = spyOn(adapter, 'iterate').and.callThrough(); + + const latestEntry = await svc.getLatestFullStateOpEntry(); + expect(latestEntry?.op.id).toBe(latestImport.id); + expect(latestEntry?.source).toBe('remote'); + expect(iterateSpy).not.toHaveBeenCalled(); + }); + + it('rebuilds the pointer on read when the meta row is absent', async () => { + const { svc, adapter } = await freshSqliteService(); + await svc.append( + createTestOperation({ id: '01900000-0000-7000-8000-000000000012' }), + ); + const latestImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000013', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await svc.append(latestImport); + + // Simulate the SQLite/migration state where the pointer was never seeded + // (the IndexedDB-only upgrade populate doesn't run on this backend). + await adapter.delete(STORE_NAMES.META, FULL_STATE_OPS_META_KEY); + + const iterateSpy = spyOn(adapter, 'iterate').and.callThrough(); + expect((await svc.getLatestFullStateOpEntry())?.op.id).toBe(latestImport.id); + expect(iterateSpy).toHaveBeenCalledTimes(1); + + // The rebuild persisted the pointer → the second read does not scan again. + expect((await svc.getLatestFullStateOpEntry())?.op.id).toBe(latestImport.id); + expect(iterateSpy).toHaveBeenCalledTimes(1); + }); + + it('clears full-state ops through the metadata pointer', async () => { + const { svc, adapter } = await freshSqliteService(); + await svc.append( + createTestOperation({ + id: '01900000-0000-7000-8000-000000000021', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }), + ); + await svc.append( + createTestOperation({ id: '01900000-0000-7000-8000-000000000022' }), + ); + + const iterateSpy = spyOn(adapter, 'iterate').and.callThrough(); + + expect(await svc.clearFullStateOps()).toBe(1); + expect(await svc.getLatestFullStateOpEntry()).toBeUndefined(); + expect((await svc.getOpsAfterSeq(0)).map((entry) => entry.op.id)).toEqual([ + '01900000-0000-7000-8000-000000000022', + ]); + expect(iterateSpy).not.toHaveBeenCalled(); + }); + }); + describe('state cache', () => { it('should save and load state cache', async () => { const testState = { task: { ids: ['1'], entities: { id1: { id: '1' } } } }; @@ -1892,6 +2209,79 @@ describe('OperationLogStoreService', () => { expect(await db.get(STORE_NAMES.CLIENT_ID, SINGLETON_KEY)).toBe('priorClient'); expect(clearCacheSpy).not.toHaveBeenCalled(); }); + + it('should update full-state metadata for the replacement op', async () => { + const syncImportOp = createTestOperation({ + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + clientId: 'replacementClient', + vectorClock: { replacementClient: 1 }, + payload: { task: { ids: [], entities: {} } }, + }); + + await service.runDestructiveStateReplacement({ + syncImportOp, + snapshotEntityKeys: [], + }); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(syncImportOp.id); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); + + it('should preserve full-state metadata when the destructive tx aborts', async () => { + const priorImport = createTestOperation({ + id: '01900000-0000-7000-8000-000000000031', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }); + await service.append(priorImport); + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(priorImport.id); + + const db = (service as any).db; + const realTransaction = db.transaction.bind(db); + spyOn(db, 'transaction').and.callFake((stores: any, mode: any) => { + const tx = realTransaction(stores, mode); + if (Array.isArray(stores) && stores.includes(STORE_NAMES.OPS)) { + const opsStore = tx.objectStore(STORE_NAMES.OPS); + opsStore.add = async () => { + throw new Error('Simulated interrupt inside destructive tx'); + }; + } + return tx; + }); + + await expectAsync( + service.runDestructiveStateReplacement({ + syncImportOp: createTestOperation({ + id: '01900000-0000-7000-8000-000000000032', + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + clientId: 'abortClient', + vectorClock: { abortClient: 1 }, + }), + snapshotEntityKeys: [], + }), + ).toBeRejected(); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + expect((await service.getLatestFullStateOpEntry())?.op.id).toBe(priorImport.id); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); }); describe('appendWithVectorClockUpdate', () => { @@ -2120,6 +2510,29 @@ describe('OperationLogStoreService', () => { expect(unsynced.length).toBe(0); }); + it('should clear full-state metadata after clearing operations', async () => { + await service.append( + createTestOperation({ + opType: OpType.SyncImport, + entityType: 'ALL' as EntityType, + entityId: undefined, + }), + ); + expect(await service.getLatestFullStateOpEntry()).toBeDefined(); + + await service.clearAllOperations(); + + const adapter = ( + service as unknown as { + _adapter: OpLogDbAdapter; + } + )._adapter; + spyOn(adapter, 'iterate').and.callThrough(); + + expect(await service.getLatestFullStateOpEntry()).toBeUndefined(); + expect(adapter.iterate).not.toHaveBeenCalled(); + }); + it('should not affect state_cache', async () => { // Save a state cache const stateCache = { 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 5223d89a62..bd4fa7d932 100644 --- a/src/app/op-log/persistence/operation-log-store.service.ts +++ b/src/app/op-log/persistence/operation-log-store.service.ts @@ -23,17 +23,23 @@ import { STORE_NAMES, SINGLETON_KEY, BACKUP_KEY, + FULL_STATE_OPS_META_KEY, OPS_INDEXES, ArchiveStoreEntry, ProfileDataStoreEntry, } from './db-keys.const'; +import { + buildFullStateOpsMeta, + FullStateOpRef, + FullStateOpsMetaEntry, +} from './full-state-ops-meta'; import { DUPLICATE_OPERATION_ERROR_MSG, OPERATION_LOG_STORE_NOT_INITIALIZED, isLockRelatedIdbOpenError, } from './op-log-errors.const'; import { runDbUpgrade } from './db-upgrade'; -import { OpLogDbAdapter } from './op-log-db-adapter'; +import { OpLogDbAdapter, OpLogTx } from './op-log-db-adapter'; import { OP_LOG_DB_ADAPTER_FACTORY } from './op-log-db-adapter.token'; import { Log } from '../../core/log'; import { @@ -114,6 +120,9 @@ const getOpId = (op: Operation | CompactOperation): string => { return op.id; }; +const getStoredOpType = (op: Operation | CompactOperation): string => + isCompactOperation(op) ? op.o : op.opType; + // Note: DBSchema requires literal string keys matching STORE_NAMES values interface OpLogDB extends DBSchema { [STORE_NAMES.OPS]: { @@ -189,6 +198,14 @@ interface OpLogDB extends DBSchema { key: string; // SINGLETON_KEY ('current') value: string; // the clientId }; + /** + * Stores small derived metadata records. Full-state op refs live here so sync + * filtering does not need to scan and decode the full ops table every call. + */ + [STORE_NAMES.META]: { + key: string; + value: FullStateOpsMetaEntry; + }; } type OpLogStoreName = (typeof STORE_NAMES)[keyof typeof STORE_NAMES]; @@ -413,6 +430,131 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort r.opId !== ref.opId), ref]; + return buildFullStateOpsMeta(refs); + } + + private _withoutFullStateRefs( + meta: FullStateOpsMetaEntry | undefined, + opIdsToRemove: Set, + ): FullStateOpsMetaEntry { + const refs = (meta?.refs ?? []).filter((ref) => !opIdsToRemove.has(ref.opId)); + return buildFullStateOpsMeta(refs); + } + + private async _rebuildFullStateOpsMetaInTx( + tx: OpLogTx, + ): Promise { + const refs: FullStateOpRef[] = []; + await tx.iterate(STORE_NAMES.OPS, {}, (value, key) => { + const ref = this._getFullStateRef(value.op, key as number); + if (ref) { + refs.push(ref); + } + return 'continue'; + }); + + const meta = buildFullStateOpsMeta(refs); + await tx.put(STORE_NAMES.META, meta, FULL_STATE_OPS_META_KEY); + return meta; + } + + private async _getFullStateOpsMetaInTxOrRebuild( + tx: OpLogTx, + ): Promise { + const meta = this._normalizeFullStateOpsMeta( + await tx.get(STORE_NAMES.META, FULL_STATE_OPS_META_KEY), + ); + return meta ?? (await this._rebuildFullStateOpsMetaInTx(tx)); + } + + private async _recordFullStateOpInTx( + tx: OpLogTx, + op: Operation | CompactOperation, + seq: number, + ): Promise { + const ref = this._getFullStateRef(op, seq); + if (!ref) { + return; + } + + const meta = await this._getFullStateOpsMetaInTxOrRebuild(tx); + await tx.put( + STORE_NAMES.META, + this._withFullStateRef(meta, ref), + FULL_STATE_OPS_META_KEY, + ); + } + + private async _rebuildFullStateOpsMeta(): Promise { + const refs: FullStateOpRef[] = []; + await this._adapter.iterate( + STORE_NAMES.OPS, + { mode: 'readonly' }, + (value, key) => { + const ref = this._getFullStateRef(value.op, key as number); + if (ref) { + refs.push(ref); + } + return 'continue'; + }, + ); + + const meta = buildFullStateOpsMeta(refs); + await this._adapter.put(STORE_NAMES.META, meta, FULL_STATE_OPS_META_KEY); + return meta; + } + + private async _getFullStateOpsMetaOrRebuild(): Promise { + return ( + this._normalizeFullStateOpsMeta( + await this._adapter.get(STORE_NAMES.META, FULL_STATE_OPS_META_KEY), + ) ?? (await this._rebuildFullStateOpsMeta()) + ); + } + async append( op: Operation, source: 'local' | 'remote' = 'local', @@ -420,6 +562,18 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); try { + if (isFullStateOpType(op.opType)) { + return await this._adapter.transaction( + [STORE_NAMES.OPS, STORE_NAMES.META], + 'readwrite', + async (tx) => { + const entry = this._buildStoredEntry(op, source, options); + const seq = await tx.add(STORE_NAMES.OPS, entry); + await this._recordFullStateOpInTx(tx, entry.op, seq); + return seq; + }, + ); + } return await this._adapter.add( STORE_NAMES.OPS, this._buildStoredEntry(op, source, options), @@ -436,21 +590,20 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); try { - return await this._adapter.transaction( - [STORE_NAMES.OPS], - 'readwrite', - async (tx) => { - const seqs: number[] = []; - for (const op of ops) { - const seq = await tx.add( - STORE_NAMES.OPS, - this._buildStoredEntry(op, source, options), - ); - seqs.push(seq); - } - return seqs; - }, - ); + const storeNames: OpLogStoreName[] = [STORE_NAMES.OPS]; + if (ops.some((op) => isFullStateOpType(op.opType))) { + storeNames.push(STORE_NAMES.META); + } + return await this._adapter.transaction(storeNames, 'readwrite', async (tx) => { + const seqs: number[] = []; + for (const op of ops) { + const entry = this._buildStoredEntry(op, source, options); + const seq = await tx.add(STORE_NAMES.OPS, entry); + await this._recordFullStateOpInTx(tx, entry.op, seq); + seqs.push(seq); + } + return seqs; + }); } catch (e) { this._handleAppendError(e); } @@ -485,7 +638,12 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + const storeNames: OpLogStoreName[] = [STORE_NAMES.OPS]; + if (ops.some((op) => isFullStateOpType(op.opType))) { + storeNames.push(STORE_NAMES.META); + } + + await this._adapter.transaction(storeNames, 'readwrite', async (tx) => { for (const op of ops) { // Check if op already exists in the same transaction (atomic) const existingKey = await tx.getKeyFromIndex( @@ -498,10 +656,9 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { - let latestEntry: OperationLogEntry | undefined; - - await this._forEachFullStateOp((entry) => { - // Track the latest by UUIDv7 (lexicographic comparison works for UUIDv7). - // We never stop early: UUIDv7 order can differ from seq order when remote - // ops with earlier timestamps arrive later, so we must scan all full-state - // ops to find the one with the latest UUIDv7 id. - if (!latestEntry || entry.op.id > latestEntry.op.id) { - latestEntry = entry; - } - return 'continue'; - }); - - return latestEntry; - } - - /** - * Iterates all ops, decodes each entry, and invokes `cb` for every - * full-state op (SYNC_IMPORT, BACKUP_IMPORT, REPAIR). Return `'stop'` - * from `cb` to end the scan early. - * - * Shared by `getLatestFullStateOpEntry` and `clearFullStateOpsExcept` - * to avoid duplicating the decode + isFullStateOpType check. - */ - private async _forEachFullStateOp( - cb: (entry: OperationLogEntry) => 'continue' | 'stop', - ): Promise { await this._ensureInit(); - await this._adapter.iterate( + + const meta = await this._getFullStateOpsMetaOrRebuild(); + if (!meta.latest) { + return undefined; + } + + const stored = await this._adapter.get( STORE_NAMES.OPS, - { mode: 'readonly' }, - (value) => { - const entry = decodeStoredEntry(value); - if (isFullStateOpType(entry.op.opType)) { - return cb(entry); - } - return 'continue'; - }, + meta.latest.seq, ); + if ( + stored && + getOpId(stored.op) === meta.latest.opId && + isFullStateOpType(getStoredOpType(stored.op)) + ) { + return decodeStoredEntry(stored); + } + + const rebuiltMeta = await this._rebuildFullStateOpsMeta(); + if (!rebuiltMeta.latest) { + return undefined; + } + const rebuiltStored = await this._adapter.get( + STORE_NAMES.OPS, + rebuiltMeta.latest.seq, + ); + return rebuiltStored ? decodeStoredEntry(rebuiltStored) : undefined; } /** @@ -709,28 +854,6 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { - if (ids.length === 0) { - return; - } - await this._adapter.transaction([STORE_NAMES.OPS], 'readwrite', async (tx) => { - for (const id of ids) { - await tx.iterate( - STORE_NAMES.OPS, - { index: OPS_INDEXES.BY_ID, query: id }, - () => 'delete-stop', - ); - } - }); - this._invalidateUnsyncedCache(); - } - /** * Deletes all full-state operations (SYNC_IMPORT, BACKUP_IMPORT, REPAIR) from the local store, * EXCEPT for the operation(s) with the specified ID(s). @@ -749,18 +872,47 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { + await this._ensureInit(); + const excludeIdSet = new Set(excludeIds); - const opsToDelete: string[] = []; + let deletedCount = 0; + await this._adapter.transaction( + [STORE_NAMES.OPS, STORE_NAMES.META], + 'readwrite', + async (tx) => { + // Read meta INSIDE the tx so a full-state append committed between the + // read and the write can't be clobbered by a stale snapshot. The + // OPS deletes and the META update then stay atomic, matching + // deleteOpsWhere — no reliance on the OPERATION_LOG lock for safety. + const meta = await this._getFullStateOpsMetaInTxOrRebuild(tx); + const refsToDelete = meta.refs.filter((ref) => !excludeIdSet.has(ref.opId)); + if (refsToDelete.length === 0) { + return; + } - await this._forEachFullStateOp((entry) => { - if (!excludeIdSet.has(entry.op.id)) { - opsToDelete.push(entry.op.id); - } - return 'continue'; - }); - - await this._deleteOpsByIds(opsToDelete); - return opsToDelete.length; + const opIdsToDelete = new Set(refsToDelete.map((ref) => ref.opId)); + for (const ref of refsToDelete) { + const stored = await tx.get(STORE_NAMES.OPS, ref.seq); + if ( + stored && + getOpId(stored.op) === ref.opId && + isFullStateOpType(getStoredOpType(stored.op)) + ) { + await tx.delete(STORE_NAMES.OPS, ref.seq); + deletedCount++; + } + } + await tx.put( + STORE_NAMES.META, + this._withoutFullStateRefs(meta, opIdsToDelete), + FULL_STATE_OPS_META_KEY, + ); + }, + ); + if (deletedCount > 0) { + this._invalidateUnsyncedCache(); + } + return deletedCount; } async getUnsynced(): Promise { @@ -976,15 +1128,34 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort(STORE_NAMES.OPS, {}, (value) => { - // Decode stored entry before applying predicate - const decoded = decodeStoredEntry(value); - if (predicate(decoded)) { - deletedCount++; - return 'delete'; - } - return 'continue'; - }); + const deletedFullStateOpIds = new Set(); + await this._adapter.transaction( + [STORE_NAMES.OPS, STORE_NAMES.META], + 'readwrite', + async (tx) => { + await tx.iterate(STORE_NAMES.OPS, {}, (value) => { + // Decode stored entry before applying predicate + const decoded = decodeStoredEntry(value); + if (predicate(decoded)) { + deletedCount++; + if (isFullStateOpType(decoded.op.opType)) { + deletedFullStateOpIds.add(decoded.op.id); + } + return 'delete'; + } + return 'continue'; + }); + + if (deletedFullStateOpIds.size > 0) { + const meta = await this._getFullStateOpsMetaInTxOrRebuild(tx); + await tx.put( + STORE_NAMES.META, + this._withoutFullStateRefs(meta, deletedFullStateOpIds), + FULL_STATE_OPS_META_KEY, + ); + } + }, + ); // Invalidate caches if any ops were deleted to prevent stale data if (deletedCount > 0) { @@ -1240,6 +1411,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { for (const store of allStores) { @@ -1310,7 +1482,18 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { await this._ensureInit(); - await this._adapter.clear(STORE_NAMES.OPS); + await this._adapter.transaction( + [STORE_NAMES.OPS, STORE_NAMES.META], + 'readwrite', + async (tx) => { + await tx.clear(STORE_NAMES.OPS); + await tx.put( + STORE_NAMES.META, + buildFullStateOpsMeta([]), + FULL_STATE_OPS_META_KEY, + ); + }, + ); this._invalidateAppliedAndUnsyncedCaches(); } @@ -1546,32 +1729,31 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort { - // 1. Append operation to ops store (encoded to compact format) - const seq = await tx.add( - STORE_NAMES.OPS, - this._buildStoredEntry(op, source, options), + const storeNames: OpLogStoreName[] = [STORE_NAMES.OPS, STORE_NAMES.VECTOR_CLOCK]; + if (isFullStateOpType(op.opType)) { + storeNames.push(STORE_NAMES.META); + } + return await this._adapter.transaction(storeNames, 'readwrite', async (tx) => { + // 1. Append operation to ops store (encoded to compact format) + const entry = this._buildStoredEntry(op, source, options); + const seq = await tx.add(STORE_NAMES.OPS, entry); + await this._recordFullStateOpInTx(tx, entry.op, seq); + + // 2. Update vector clock to match the operation's clock (only for + // local ops). The op.vectorClock already contains the incremented + // value from the caller; we store it as the current clock so + // subsequent operations can build on it. + if (source === 'local') { + await tx.put( + STORE_NAMES.VECTOR_CLOCK, + { clock: op.vectorClock, lastUpdate: Date.now() }, + SINGLETON_KEY, ); + this._vectorClockCache = op.vectorClock; + } - // 2. Update vector clock to match the operation's clock (only for - // local ops). The op.vectorClock already contains the incremented - // value from the caller; we store it as the current clock so - // subsequent operations can build on it. - if (source === 'local') { - await tx.put( - STORE_NAMES.VECTOR_CLOCK, - { clock: op.vectorClock, lastUpdate: Date.now() }, - SINGLETON_KEY, - ); - this._vectorClockCache = op.vectorClock; - } - - return seq; - }, - ); + return seq; + }); } catch (e) { this._handleAppendError(e); } @@ -1617,6 +1799,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort