mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
perf(sync): cache latest full-state op lookup (#8427)
* 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 <johannes.millan@gmail.com>
This commit is contained in:
Symon Baikov
GitHub
parent
d50c74eda7
commit
807d3bf5e5
7 changed files with 949 additions and 139 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> =>
|
||||
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<any, any, 'versionchange'>,
|
||||
): 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);
|
||||
}
|
||||
};
|
||||
|
|
|
|||
48
src/app/op-log/persistence/full-state-ops-meta.ts
Normal file
48
src/app/op-log/persistence/full-state-ops-meta.ts
Normal file
|
|
@ -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>,
|
||||
): FullStateOpRef | undefined =>
|
||||
refs.reduce<FullStateOpRef | undefined>(
|
||||
(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<FullStateOpRef>,
|
||||
): FullStateOpsMetaEntry => {
|
||||
const latest = getLatestFullStateRef(refs);
|
||||
return latest ? { refs: [...refs], latest } : { refs: [...refs] };
|
||||
};
|
||||
|
|
@ -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 },
|
||||
],
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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<unknown>;
|
||||
}
|
||||
).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<unknown>;
|
||||
}
|
||||
).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<unknown>;
|
||||
}
|
||||
).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 = {
|
||||
|
|
|
|||
|
|
@ -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<O
|
|||
this._invalidateUnsyncedCache();
|
||||
}
|
||||
|
||||
private _getFullStateRef(
|
||||
op: Operation | CompactOperation,
|
||||
seq: number,
|
||||
): FullStateOpRef | undefined {
|
||||
return isFullStateOpType(getStoredOpType(op))
|
||||
? { opId: getOpId(op), seq }
|
||||
: undefined;
|
||||
}
|
||||
|
||||
private _normalizeFullStateOpsMeta(meta: unknown): FullStateOpsMetaEntry | undefined {
|
||||
if (typeof meta !== 'object' || meta === null || !('refs' in meta)) {
|
||||
return undefined;
|
||||
}
|
||||
const refs = (meta as { refs: unknown }).refs;
|
||||
if (!Array.isArray(refs)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const normalizedRefs: FullStateOpRef[] = [];
|
||||
for (const ref of refs) {
|
||||
if (
|
||||
typeof ref !== 'object' ||
|
||||
ref === null ||
|
||||
!('opId' in ref) ||
|
||||
!('seq' in ref)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const { opId, seq } = ref as { opId: unknown; seq: unknown };
|
||||
if (typeof opId !== 'string' || typeof seq !== 'number') {
|
||||
return undefined;
|
||||
}
|
||||
normalizedRefs.push({ opId, seq });
|
||||
}
|
||||
|
||||
return buildFullStateOpsMeta(normalizedRefs);
|
||||
}
|
||||
|
||||
private _withFullStateRef(
|
||||
meta: FullStateOpsMetaEntry | undefined,
|
||||
ref: FullStateOpRef,
|
||||
): FullStateOpsMetaEntry {
|
||||
const refs = [...(meta?.refs ?? []).filter((r) => r.opId !== ref.opId), ref];
|
||||
return buildFullStateOpsMeta(refs);
|
||||
}
|
||||
|
||||
private _withoutFullStateRefs(
|
||||
meta: FullStateOpsMetaEntry | undefined,
|
||||
opIdsToRemove: Set<string>,
|
||||
): FullStateOpsMetaEntry {
|
||||
const refs = (meta?.refs ?? []).filter((ref) => !opIdsToRemove.has(ref.opId));
|
||||
return buildFullStateOpsMeta(refs);
|
||||
}
|
||||
|
||||
private async _rebuildFullStateOpsMetaInTx(
|
||||
tx: OpLogTx,
|
||||
): Promise<FullStateOpsMetaEntry> {
|
||||
const refs: FullStateOpRef[] = [];
|
||||
await tx.iterate<StoredOperationLogEntry>(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<FullStateOpsMetaEntry> {
|
||||
const meta = this._normalizeFullStateOpsMeta(
|
||||
await tx.get<unknown>(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<void> {
|
||||
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<FullStateOpsMetaEntry> {
|
||||
const refs: FullStateOpRef[] = [];
|
||||
await this._adapter.iterate<StoredOperationLogEntry>(
|
||||
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<FullStateOpsMetaEntry> {
|
||||
return (
|
||||
this._normalizeFullStateOpsMeta(
|
||||
await this._adapter.get<unknown>(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<O
|
|||
): Promise<number> {
|
||||
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<O
|
|||
): Promise<number[]> {
|
||||
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<O
|
|||
const writtenOps: Operation[] = [];
|
||||
let skippedCount = 0;
|
||||
|
||||
await this._adapter.transaction([STORE_NAMES.OPS], 'readwrite', async (tx) => {
|
||||
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<O
|
|||
continue;
|
||||
}
|
||||
|
||||
const seq = await tx.add(
|
||||
STORE_NAMES.OPS,
|
||||
this._buildStoredEntry(op, source, options),
|
||||
);
|
||||
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);
|
||||
writtenOps.push(op);
|
||||
}
|
||||
|
|
@ -647,52 +804,40 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
* - Local unsynced imports (source='local', no syncedAt) → show dialog
|
||||
* - Remote/synced imports → silently filter old ops (already accepted)
|
||||
*
|
||||
* Uses cursor iteration for memory efficiency - avoids loading all ops into memory
|
||||
* at once, which matters on mobile devices with limited RAM. Trade-off is slightly
|
||||
* slower due to per-op cursor overhead vs bulk getAll().
|
||||
* Uses the persistent full-state metadata pointer. Existing databases rebuild
|
||||
* that metadata once on first read, then future calls are O(1).
|
||||
*
|
||||
* @returns The latest full-state operation entry, or undefined if none exists
|
||||
*/
|
||||
async getLatestFullStateOpEntry(): Promise<OperationLogEntry | undefined> {
|
||||
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<void> {
|
||||
await this._ensureInit();
|
||||
await this._adapter.iterate<StoredOperationLogEntry>(
|
||||
|
||||
const meta = await this._getFullStateOpsMetaOrRebuild();
|
||||
if (!meta.latest) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const stored = await this._adapter.get<StoredOperationLogEntry>(
|
||||
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<StoredOperationLogEntry>(
|
||||
STORE_NAMES.OPS,
|
||||
rebuiltMeta.latest.seq,
|
||||
);
|
||||
return rebuiltStored ? decodeStoredEntry(rebuiltStored) : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -709,28 +854,6 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
return this.clearFullStateOpsExcept([]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Deletes ops by their `op.id` via the unique byId index, atomically.
|
||||
* Mirrors the original keyed-index-cursor delete used by the
|
||||
* clearFullStateOps* methods. No-op (and no cache invalidation) for an
|
||||
* empty list, matching prior behavior.
|
||||
*/
|
||||
private async _deleteOpsByIds(ids: string[]): Promise<void> {
|
||||
if (ids.length === 0) {
|
||||
return;
|
||||
}
|
||||
await this._adapter.transaction([STORE_NAMES.OPS], 'readwrite', async (tx) => {
|
||||
for (const id of ids) {
|
||||
await tx.iterate<StoredOperationLogEntry>(
|
||||
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<O
|
|||
* @returns Number of operations deleted
|
||||
*/
|
||||
async clearFullStateOpsExcept(excludeIds: string[]): Promise<number> {
|
||||
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<StoredOperationLogEntry>(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<OperationLogEntry[]> {
|
||||
|
|
@ -976,15 +1128,34 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
// Iterate the whole store, deleting entries that match the predicate.
|
||||
// (A range delete isn't possible — the predicate is on decoded fields.)
|
||||
let deletedCount = 0;
|
||||
await this._adapter.iterate<StoredOperationLogEntry>(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<string>();
|
||||
await this._adapter.transaction(
|
||||
[STORE_NAMES.OPS, STORE_NAMES.META],
|
||||
'readwrite',
|
||||
async (tx) => {
|
||||
await tx.iterate<StoredOperationLogEntry>(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<O
|
|||
STORE_NAMES.ARCHIVE_OLD,
|
||||
STORE_NAMES.PROFILE_DATA,
|
||||
STORE_NAMES.CLIENT_ID,
|
||||
STORE_NAMES.META,
|
||||
];
|
||||
await this._adapter.transaction(allStores, 'readwrite', async (tx) => {
|
||||
for (const store of allStores) {
|
||||
|
|
@ -1310,7 +1482,18 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
*/
|
||||
async clearAllOperations(): Promise<void> {
|
||||
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<O
|
|||
await this._ensureInit();
|
||||
|
||||
try {
|
||||
return await this._adapter.transaction(
|
||||
[STORE_NAMES.OPS, STORE_NAMES.VECTOR_CLOCK],
|
||||
'readwrite',
|
||||
async (tx) => {
|
||||
// 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<O
|
|||
// Unconditional: both callers (clean-slate, backup-restore) always rotate
|
||||
// the clientId. Unlike the archive stores it is never conditional.
|
||||
STORE_NAMES.CLIENT_ID,
|
||||
STORE_NAMES.META,
|
||||
];
|
||||
if (archiveYoung != null) {
|
||||
storeNames.push(STORE_NAMES.ARCHIVE_YOUNG);
|
||||
|
|
@ -1645,6 +1828,14 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
|
|||
STORE_NAMES.OPS,
|
||||
this._buildStoredEntry(syncImportOp, 'local'),
|
||||
);
|
||||
// syncImportOp is always a full-state op (both callers pass SYNC_IMPORT);
|
||||
// OPS was just cleared, so the pointer is exactly this one op. Use the
|
||||
// shared builder so `latest` is derived, never hand-asserted.
|
||||
await tx.put(
|
||||
STORE_NAMES.META,
|
||||
buildFullStateOpsMeta([{ opId: syncImportOp.id, seq }]),
|
||||
FULL_STATE_OPS_META_KEY,
|
||||
);
|
||||
|
||||
await tx.put(
|
||||
STORE_NAMES.VECTOR_CLOCK,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue