diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.ts index e4d75a20..8ea1e006 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.ts @@ -44,11 +44,16 @@ type ReplicateResultProcessorServices = Pick< * awaiting it. Result application can still be running inside work admitted by * that owner, so awaiting retirement here could make each side wait for the * other to finish. + * + * Runtime databases are deliberately obtained through operation-time + * accessors. Feature composition precedes their initialisation, and database + * reset may replace their backing instances, so retaining an earlier concrete + * database would be invalid. */ interface ReplicateResultProcessorContext { readonly currentSettings: () => ReplicateResultProcessorSettings; - readonly keyValueDB: LiveSyncBaseCore["kvDB"]; - readonly localDatabase: LiveSyncBaseCore["localDatabase"]; + readonly getKeyValueDB: () => LiveSyncBaseCore["kvDB"]; + readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"]; readonly requestActiveReplicatorRetirement: () => void; readonly runLocalApplicationActivity: ( task: () => T | PromiseLike, @@ -77,7 +82,7 @@ export class ReplicateResultProcessor { constructor(private readonly context: ReplicateResultProcessorContext) {} private get localDatabase() { - return this.context.localDatabase; + return this.context.getLocalDatabase(); } private get services() { return this.context.services; @@ -119,7 +124,7 @@ export class ReplicateResultProcessor { queued: this._queuedChanges.slice(), processing: this._processingChanges.slice(), } satisfies ReplicateResultProcessorState; - await this.context.keyValueDB.set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot); + await this.context.getKeyValueDB().set(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT, snapshot); this.log( `Snapshot taken. Queued: ${snapshot.queued.length}, Processing: ${snapshot.processing.length}`, LOG_LEVEL_DEBUG @@ -141,9 +146,9 @@ export class ReplicateResultProcessor { * Restore from snapshot. */ public async restoreFromSnapshot() { - const snapshot = await this.context.keyValueDB.get( - KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT - ); + const snapshot = await this.context + .getKeyValueDB() + .get(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT); if (snapshot) { // Restoring the snapshot re-runs processing for both queued and processing items. const newQueue = [...snapshot.processing, ...snapshot.queued, ...this._queuedChanges]; diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts index 08225a37..c59e11be 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts @@ -56,8 +56,8 @@ function setup(options: SetupOptions = {}) { }; const processor = new ReplicateResultProcessor({ currentSettings: () => ({ maxMTimeForReflectEvents: 0, suspendParseReplicationResult: false }), - keyValueDB: core.kvDB, - localDatabase: core.localDatabase, + getKeyValueDB: () => core.kvDB, + getLocalDatabase: () => core.localDatabase, requestActiveReplicatorRetirement: () => { void onCloseActiveReplication(); }, @@ -95,14 +95,16 @@ describe("ReplicateResultProcessor", () => { const findAllNormalDocs = vi.fn(async function* () { yield* documents; }); + const getLocalDatabase = vi.fn(() => ({ findAllNormalDocs })); const processor = new ReplicateResultProcessor({ - localDatabase: { findAllNormalDocs }, + getLocalDatabase, } as never); const enqueueAll = vi.spyOn(processor, "enqueueAll").mockImplementation(() => undefined); await expect(processor.reprocessStoredDocuments()).resolves.toBe(2); expect(findAllNormalDocs).toHaveBeenCalledOnce(); + expect(getLocalDatabase).toHaveBeenCalledOnce(); expect(enqueueAll).toHaveBeenCalledOnce(); expect(enqueueAll).toHaveBeenCalledWith(documents); }); diff --git a/src/serviceFeatures/replication/centralCompatibilityRecovery.ts b/src/serviceFeatures/replication/centralCompatibilityRecovery.ts index 648a5f24..47257d5f 100644 --- a/src/serviceFeatures/replication/centralCompatibilityRecovery.ts +++ b/src/serviceFeatures/replication/centralCompatibilityRecovery.ts @@ -17,9 +17,11 @@ type CentralCompatibilityRecoveryServices = Pick< "API" | "appLifecycle" | "replicator" | "tweakValue" >; +/** Collaborators for applying a compatibility decision to its failed publication. */ interface CentralCompatibilityRecoveryContext { readonly confirm: LiveSyncBaseCore["confirm"]; - readonly localDatabase: LiveSyncBaseCore["localDatabase"]; + /** Obtain the database only when recovery runs, after initialisation or reset. */ + readonly getLocalDatabase: () => LiveSyncBaseCore["localDatabase"]; readonly rebuilder: LiveSyncBaseCore["rebuilder"]; readonly services: CentralCompatibilityRecoveryServices; } @@ -56,7 +58,7 @@ export function createCentralCompatibilityRecovery(context: CentralCompatibility ) { Logger("The remote database has been cleaned.", showMessage ? LOG_LEVEL_NOTICE : LOG_LEVEL_INFO); await skipIfDuplicated("cleanup", async () => { - const count = await purgeUnreferencedChunks(context.localDatabase.localDatabase, true); + const count = await purgeUnreferencedChunks(context.getLocalDatabase().localDatabase, true); const message = `The remote database has been cleaned up. To synchronize, this device must be also cleaned up. ${count} chunk(s) will be erased from this device. However, If there are many chunks to be deleted, maybe fetching again is faster. @@ -83,6 +85,7 @@ Even if you choose to clean up, you will see this option again if you exit Obsid if (activeContext !== expectedContext) return; const replicator = activeContext.replicator; if (!(replicator instanceof LiveSyncCouchDBReplicator)) return; + const localDatabase = context.getLocalDatabase(); const remoteDatabase = await replicator.connectRemoteCouchDBWithSetting( setting, context.services.API.isMobile(), @@ -94,16 +97,16 @@ Even if you choose to clean up, you will see this option again if you exit Obsid } try { - await purgeUnreferencedChunks(context.localDatabase.localDatabase, false); - context.localDatabase.clearCaches(); + await purgeUnreferencedChunks(localDatabase.localDatabase, false); + localDatabase.clearCaches(); const replicated = await context.services.replicator.runFiniteReplicationActivity( () => replicator.openOneShotReplication(setting, showMessage, false, "sync", true), { label: "replication" } ); if (replicated) { - await balanceChunkPurgedDBs(context.localDatabase.localDatabase, remoteDatabase.db); - await purgeUnreferencedChunks(context.localDatabase.localDatabase, false); - context.localDatabase.clearCaches(); + await balanceChunkPurgedDBs(localDatabase.localDatabase, remoteDatabase.db); + await purgeUnreferencedChunks(localDatabase.localDatabase, false); + localDatabase.clearCaches(); await replicator.markRemoteResolved(setting); Logger( "The local database has been cleaned up.", diff --git a/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts b/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts index 0ef47748..50d2714f 100644 --- a/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts +++ b/src/serviceFeatures/replication/centralCompatibilityRecovery.unit.spec.ts @@ -28,7 +28,7 @@ describe("central compatibility recovery", () => { try { const recovery = createCentralCompatibilityRecovery({ confirm: {}, - localDatabase: {}, + getLocalDatabase: () => ({}), rebuilder: {}, services: { appLifecycle: {}, @@ -75,7 +75,7 @@ describe("central compatibility recovery", () => { }); const recovery = createCentralCompatibilityRecovery({ confirm: {}, - localDatabase: {}, + getLocalDatabase: () => ({}), rebuilder: {}, services: { appLifecycle: {}, @@ -137,7 +137,7 @@ describe("central compatibility recovery", () => { ); const recovery = createCentralCompatibilityRecovery({ confirm: {}, - localDatabase: {}, + getLocalDatabase: () => ({}), rebuilder: {}, services: { appLifecycle: {}, @@ -180,7 +180,7 @@ describe("central compatibility recovery", () => { confirm: { askSelectStringDialogue: vi.fn(async (_message: string, choices: string[]) => choices[1]), }, - localDatabase: {}, + getLocalDatabase: () => ({}), rebuilder: {}, services: { appLifecycle: { scheduleRestart: vi.fn() }, @@ -228,9 +228,10 @@ describe("central compatibility recovery", () => { task(expectedContext) ); const localDatabase = { localDatabase: {}, clearCaches: vi.fn() }; + const getLocalDatabase = vi.fn(() => localDatabase); const recovery = createCentralCompatibilityRecovery({ confirm: { confirmWithMessage: vi.fn(async () => "Cleanup") }, - localDatabase, + getLocalDatabase, rebuilder: {}, services: { appLifecycle: {}, @@ -258,6 +259,7 @@ describe("central compatibility recovery", () => { activityFinished.mock.invocationCallOrder[0] ); expect(chunkMocks.balanceChunkPurgedDBs).toHaveBeenCalledOnce(); + expect(getLocalDatabase).toHaveBeenCalledTimes(2); expect(close).toHaveBeenCalledOnce(); expect(close.mock.invocationCallOrder[0]).toBeLessThan(activityFinished.mock.invocationCallOrder[0]); }); diff --git a/src/serviceFeatures/replication/index.ts b/src/serviceFeatures/replication/index.ts index 813f106a..b29da6e4 100644 --- a/src/serviceFeatures/replication/index.ts +++ b/src/serviceFeatures/replication/index.ts @@ -39,8 +39,8 @@ export function useReplicationFeature services.setting.currentSettings(), - keyValueDB: services.keyValueDB.kvDB, - localDatabase: core.localDatabase, + getKeyValueDB: () => services.keyValueDB.kvDB, + getLocalDatabase: () => core.localDatabase, requestActiveReplicatorRetirement: () => { // Do not await a retirement transition from result application: it // may be draining the replication work which delivered this item. @@ -81,7 +81,7 @@ export function useReplicationFeature core.localDatabase, rebuilder: core.rebuilder, services: { API: services.API, diff --git a/src/serviceFeatures/replication/replicationFeature.unit.spec.ts b/src/serviceFeatures/replication/replicationFeature.unit.spec.ts index ed4fe317..13e190d0 100644 --- a/src/serviceFeatures/replication/replicationFeature.unit.spec.ts +++ b/src/serviceFeatures/replication/replicationFeature.unit.spec.ts @@ -6,8 +6,29 @@ import { useReplicationFeature } from "./index"; type BooleanHandler = (showMessage: boolean) => Promise; type ParseHandler = (documents: PouchDB.Core.ExistingDocument[]) => Promise; +type KeyValueDBFixture = { + readonly kvDB: { + get: (key: string) => Promise; + set: (key: string, value: unknown) => Promise; + }; +}; +type SetupOptions = { + readonly getLocalDatabase?: () => object; + readonly keyValueDB?: KeyValueDBFixture; + readonly onCloseActiveReplication?: () => Promise; +}; -function setup(onCloseActiveReplication = vi.fn(async () => true)) { +function setup(options: SetupOptions = {}) { + const { + getLocalDatabase = () => ({}), + keyValueDB = { + kvDB: { + get: vi.fn(async () => undefined), + set: vi.fn(async () => undefined), + }, + }, + onCloseActiveReplication = vi.fn(async () => true), + } = options; const read = vi.fn(async () => new Uint8Array([1])); const dispose = vi.fn(async () => undefined); const createRemoteResource = vi.fn(async () => ({ read, dispose })); @@ -24,12 +45,7 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) { }, context: createServiceContext(), databaseEvents: { onDatabaseInitialised: { addHandler: vi.fn() } }, - keyValueDB: { - kvDB: { - get: vi.fn(async () => undefined), - set: vi.fn(async () => undefined), - }, - }, + keyValueDB, path: { getPath: vi.fn((entry: { path: string }) => entry.path) }, replication: { onBeforeReplicate: { @@ -59,7 +75,9 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) { }; const core = { confirm: {}, - localDatabase: {}, + get localDatabase() { + return getLocalDatabase(); + }, rebuilder: {}, services, }; @@ -80,6 +98,42 @@ function setup(onCloseActiveReplication = vi.fn(async () => true)) { } describe("replication serviceFeature composition", () => { + it("does not acquire the local database while composing result handlers", () => { + const acquireLocalDatabase = vi.fn(() => { + throw new Error("Local database is not ready yet"); + }); + + expect(() => setup({ getLocalDatabase: acquireLocalDatabase })).not.toThrow(); + expect(acquireLocalDatabase).not.toHaveBeenCalled(); + }); + + it("acquires the current key-value database only when snapshot recovery starts", async () => { + const backingDatabase = { + get: vi.fn(async () => undefined), + set: vi.fn(async () => undefined), + }; + let isReady = false; + const acquireKeyValueDB = vi.fn(() => { + if (!isReady) throw new Error("KeyValueDB is not initialized yet"); + return backingDatabase; + }); + const keyValueDB = { + get kvDB() { + return acquireKeyValueDB(); + }, + }; + + const { beforeReplicateHandlers } = setup({ keyValueDB }); + + expect(acquireKeyValueDB).not.toHaveBeenCalled(); + isReady = true; + const restoreSnapshot = beforeReplicateHandlers.get(100); + expect(restoreSnapshot).toBeDefined(); + await expect(restoreSnapshot!(false)).resolves.toBe(true); + expect(acquireKeyValueDB).toHaveBeenCalledOnce(); + expect(backingDatabase.get).toHaveBeenCalledOnce(); + }); + it("refreshes and disposes the remote Security Seed before central replication", async () => { const { centralRemoteHandlers, createRemoteResource, dispose, read } = setup(); @@ -109,7 +163,7 @@ describe("replication serviceFeature composition", () => { it("requests owner retirement without awaiting the transition from result application", async () => { const retirement = promiseWithResolvers(); const onCloseActiveReplication = vi.fn(() => retirement.promise); - const harness = setup(onCloseActiveReplication); + const harness = setup({ onCloseActiveReplication }); const versionInfo = { _id: "versioninfo", _rev: "1-test",