diff --git a/src/LiveSyncBaseCore.ts b/src/LiveSyncBaseCore.ts index 6255cedf..3ae88b72 100644 --- a/src/LiveSyncBaseCore.ts +++ b/src/LiveSyncBaseCore.ts @@ -40,6 +40,7 @@ export interface LiveSyncCoreFeatureViews { } export interface StartupDatabaseOptions { + readonly discardReplicationResultSnapshotOnReset?: boolean; readonly ignoreSuspending?: boolean; readonly continueOnFileFailure?: boolean; } diff --git a/src/apps/cli/main.bootstrap.unit.spec.ts b/src/apps/cli/main.bootstrap.unit.spec.ts index 631ba323..558d3fd3 100644 --- a/src/apps/cli/main.bootstrap.unit.spec.ts +++ b/src/apps/cli/main.bootstrap.unit.spec.ts @@ -14,6 +14,7 @@ import { type FilePathWithPrefix, type ObsidianLiveSyncSettings, } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { ReplicateResultProcessor } from "@/serviceFeatures/replication/ReplicateResultProcessor"; import { ServiceFileHandler } from "@/serviceModules/FileHandler"; import { ServiceFileAccessCLI } from "./serviceModules/ServiceFileAccessImpl"; import { runCommand } from "./commands/runCommand"; @@ -103,6 +104,39 @@ describe("CLI database preparation", () => { await expect(main(standardIo, runner)).rejects.toThrow(`__EXIT__:${exitCode}`); } + it.each(["fetch", "rebuild"] as const)("does not replay the old reflection snapshot before %s", async (command) => { + const restore = vi.spyOn(ReplicateResultProcessor.prototype, "restoreFromSnapshot"); + const runner = vi.fn(async () => false); + + await start(command, runner, 1, true); + + expect(runner).toHaveBeenCalledOnce(); + expect(restore).not.toHaveBeenCalled(); + }); + + it.each(["fetch", "rebuild"] as const)("invalidates old reflection work after the %s database reset", async (command) => { + const snapshotKey = "replicationResultProcessorSnapshot"; + const oldSnapshot = { queued: [{ _id: "old", _rev: "1-old" }], processing: [] }; + let retainedBeforeReset: unknown; + let retainedAfterReset: unknown; + let reset: boolean | undefined; + const runner = vi.fn(async (_options, { core }) => { + await core.services.keyValueDB.kvDB.set(snapshotKey, oldSnapshot); + retainedBeforeReset = await core.services.keyValueDB.kvDB.get(snapshotKey); + reset = await core.services.database.resetDatabase(); + retainedAfterReset = await core.services.keyValueDB.kvDB.get(snapshotKey); + return false; + }); + + await start(command, runner, 1, true); + + expect(runner).toHaveBeenCalledOnce(); + expect(reset).toBe(true); + expect(retainedBeforeReset).toEqual(oldSnapshot); + expect(retainedAfterReset).toEqual({ queued: [], processing: [] }); + expect(await readFile(join(vaultPath, "notes/local.md"), "utf8")).toBe("local content"); + }); + it.each([ { handleFilenameCaseSensitive: true, pathNormalizationVersion: 0 as const }, { handleFilenameCaseSensitive: true, pathNormalizationVersion: 1 as const }, diff --git a/src/apps/cli/main.ts b/src/apps/cli/main.ts index 6c93dd93..c40de3e8 100644 --- a/src/apps/cli/main.ts +++ b/src/apps/cli/main.ts @@ -77,13 +77,21 @@ const VAULT_SYNC_MODES: Readonly>> watchFiles: false, reflectReplicationResults: false, scanOnStartup: false, - startupDatabaseOptions: { ignoreSuspending: false, continueOnFileFailure: false }, + startupDatabaseOptions: { + ignoreSuspending: false, + continueOnFileFailure: false, + discardReplicationResultSnapshotOnReset: true, + }, }, fetch: { watchFiles: false, reflectReplicationResults: true, scanOnStartup: false, - startupDatabaseOptions: { ignoreSuspending: false, continueOnFileFailure: false }, + startupDatabaseOptions: { + ignoreSuspending: false, + continueOnFileFailure: false, + discardReplicationResultSnapshotOnReset: true, + }, }, }; diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.ts index a45d3658..646ee295 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.ts @@ -97,7 +97,19 @@ export class ReplicateResultProcessor { if (cause !== undefined) this.logError(cause); Logger($msg("Ui.Common.SomeFilesCouldNotBeSynchronised"), LOG_LEVEL_NOTICE); } - constructor(private readonly context: ReplicateResultProcessorContext) {} + constructor( + private readonly context: ReplicateResultProcessorContext, + private waitingForDatabaseReset = false + ) {} + + /** Discard old reflection work only after the requested startup database replacement succeeds. */ + public async completeStartupDatabaseReset(): Promise { + if (!this.waitingForDatabaseReset) return; + await this._takeSnapshot(); + this._restoreFromSnapshot = Promise.resolve(); + this.waitingForDatabaseReset = false; + this.continueHeldDocuments(); + } private get localDatabase() { return this.context.getLocalDatabase(); @@ -150,6 +162,7 @@ export class ReplicateResultProcessor { public get isSuspended() { return ( + this.waitingForDatabaseReset || this._suspended || !this.acceptsResultApplication || this.context.currentSettings().suspendParseReplicationResult || @@ -188,6 +201,7 @@ export class ReplicateResultProcessor { * Restore from snapshot. */ public async restoreFromSnapshot() { + if (this.waitingForDatabaseReset) return; const snapshot = await this.context .getKeyValueDB() .get(KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT); @@ -211,6 +225,7 @@ export class ReplicateResultProcessor { * @returns Promise that resolves when restoration is complete. */ public restoreFromSnapshotOnce() { + if (this.waitingForDatabaseReset) return Promise.resolve(); if (!this._restoreFromSnapshot) { this._restoreFromSnapshot = this.restoreFromSnapshot(); } @@ -360,6 +375,7 @@ export class ReplicateResultProcessor { * @returns */ protected enqueueChange(doc: PouchDB.Core.ExistingDocument) { + if (this.waitingForDatabaseReset) return; const old = this._queuedChanges.find((e) => e._id == doc._id); const path = "path" in doc ? this.getPath(doc) : ""; const docNote = `${path} (${shortenId(doc._id)}, ${shortenRev(doc._rev)})`; diff --git a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts index 54141175..3be9de05 100644 --- a/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts +++ b/src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts @@ -37,6 +37,7 @@ function note(id: string): PouchDB.Core.ExistingDocument { } type SetupOptions = { + waitForDatabaseReset?: boolean; onVaultReflectionFailure?: () => void; getRaw?: (id: string) => Promise; getDBEntryFromMeta?: (entry: object) => Promise; @@ -98,21 +99,24 @@ function setup(options: SetupOptions = {}) { getDBEntryFromMeta, }, }; - const processor = new ReplicateResultProcessor({ - onVaultReflectionFailure: options.onVaultReflectionFailure, - currentSettings: () => ({ - maxMTimeForReflectEvents: options.maxMTimeForReflectEvents ?? 0, - suspendParseReplicationResult: false, - pathNormalizationVersion: options.pathNormalizationVersion ?? 0, - }), - getKeyValueDB: () => core.kvDB, - getLocalDatabase: () => core.localDatabase, - requestActiveReplicatorRetirement: () => { - void onCloseActiveReplication(); - }, - runLocalApplicationActivity: runBoundedLocalApplicationActivity, - services: core.services, - } as never); + const processor = new ReplicateResultProcessor( + { + onVaultReflectionFailure: options.onVaultReflectionFailure, + currentSettings: () => ({ + maxMTimeForReflectEvents: options.maxMTimeForReflectEvents ?? 0, + suspendParseReplicationResult: false, + pathNormalizationVersion: options.pathNormalizationVersion ?? 0, + }), + getKeyValueDB: () => core.kvDB, + getLocalDatabase: () => core.localDatabase, + requestActiveReplicatorRetirement: () => { + void onCloseActiveReplication(); + }, + runLocalApplicationActivity: runBoundedLocalApplicationActivity, + services: core.services, + } as never, + options.waitForDatabaseReset + ); return { getDBEntryFromMeta, isTargetFile: core.services.vault.isTargetFile, @@ -129,6 +133,65 @@ function setup(options: SetupOptions = {}) { } describe("ReplicateResultProcessor", () => { + it("does not restore old reflection work while waiting for database replacement", async () => { + const getSnapshot = vi.fn(async () => ({ queued: [note("old")], processing: [] })); + const setSnapshot = vi.fn(async () => undefined); + const { processor, processSynchroniseResult, replicationResultCount } = setup({ + waitForDatabaseReset: true, + getSnapshot, + setSnapshot, + }); + + await processor.restoreFromSnapshotOnce(); + processor.resume(); + processor.continueHeldDocuments(); + + expect(getSnapshot).not.toHaveBeenCalled(); + expect(setSnapshot).not.toHaveBeenCalled(); + expect(replicationResultCount.value).toBe(0); + expect(processSynchroniseResult).not.toHaveBeenCalled(); + }); + + it("discards the old snapshot after reset and accepts only newly received documents", async () => { + const getSnapshot = vi.fn(async () => ({ queued: [note("old")], processing: [note("in-flight")] })); + const setSnapshot = vi.fn(async () => undefined); + const { processor, processSynchroniseResult, replicationResultCount } = setup({ + waitForDatabaseReset: true, + getSnapshot, + setSnapshot, + }); + await processor.restoreFromSnapshotOnce(); + processor.enqueueAll([note("before-reset")]); + expect(setSnapshot).not.toHaveBeenCalled(); + expect(replicationResultCount.value).toBe(0); + + await processor.completeStartupDatabaseReset(); + + expect(setSnapshot).toHaveBeenCalledWith("replicationResultProcessorSnapshot", { queued: [], processing: [] }); + await processor.restoreFromSnapshotOnce(); + expect(getSnapshot).not.toHaveBeenCalled(); + processor.enqueueAll([note("new")]); + await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce()); + expect(processSynchroniseResult).toHaveBeenCalledWith(expect.objectContaining({ path: "new.md" })); + }); + + it("keeps reflection blocked if persisting snapshot invalidation fails", async () => { + const failure = new Error("Snapshot write failed"); + const setSnapshot = vi.fn(async () => undefined).mockRejectedValueOnce(failure); + const { processor, processSynchroniseResult } = setup({ waitForDatabaseReset: true, setSnapshot }); + + await expect(processor.completeStartupDatabaseReset()).rejects.toThrow(failure); + processor.resume(); + processor.enqueueAll([note("blocked")]); + expect(processor.isSuspended).toBe(true); + expect(processSynchroniseResult).not.toHaveBeenCalled(); + + await processor.completeStartupDatabaseReset(); + processor.enqueueAll([note("after-retry")]); + await vi.waitFor(() => expect(processSynchroniseResult).toHaveBeenCalledOnce()); + expect(processSynchroniseResult).toHaveBeenCalledWith(expect.objectContaining({ path: "after-retry.md" })); + }); + it("reports an IndexedDB read error as failed reflection", async () => { const onVaultReflectionFailure = vi.fn(); const { processor, processSynchroniseResult } = setup({ diff --git a/src/serviceFeatures/replication/index.ts b/src/serviceFeatures/replication/index.ts index 884822fc..abaead26 100644 --- a/src/serviceFeatures/replication/index.ts +++ b/src/serviceFeatures/replication/index.ts @@ -37,28 +37,38 @@ export function useReplicationFeature services.context.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED), - currentSettings: () => services.setting.currentSettings(), - 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. - fireAndForget(() => services.replicator.onCloseActiveReplication()); + const discardSnapshotOnReset = core.startupDatabaseOptions?.discardReplicationResultSnapshotOnReset ?? false; + const resultProcessor = new ReplicateResultProcessor( + { + onVaultReflectionFailure: () => services.context.events.emitEvent(EVENT_VAULT_REFLECTION_FAILED), + currentSettings: () => services.setting.currentSettings(), + 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. + fireAndForget(() => services.replicator.onCloseActiveReplication()); + }, + runLocalApplicationActivity: async (task, options) => + localApplicationActivityOwner + ? await localApplicationActivityOwner.runBoundedLocalApplicationActivity(task, options) + : await task(), + services: { + appLifecycle: services.appLifecycle, + database: services.database, + path: services.path, + replication: services.replication, + vault: services.vault, + }, }, - runLocalApplicationActivity: async (task, options) => - localApplicationActivityOwner - ? await localApplicationActivityOwner.runBoundedLocalApplicationActivity(task, options) - : await task(), - services: { - appLifecycle: services.appLifecycle, - database: services.database, - path: services.path, - replication: services.replication, - vault: services.vault, - }, - }); + discardSnapshotOnReset + ); + if (discardSnapshotOnReset) { + services.database.onDatabaseReset.addHandler(async () => { + await resultProcessor.completeStartupDatabaseReset(); + return true; + }); + } const unresolvedErrorManager = new UnresolvedErrorManager(services.appLifecycle, services.context.events); const initialiseAutomaticReplicationTriggers = createAutomaticReplicationTriggers({ currentSettings: () => services.setting.currentSettings(), diff --git a/updates.md b/updates.md index 83832cb8..9d899544 100644 --- a/updates.md +++ b/updates.md @@ -48,6 +48,8 @@ Earlier releases remain available in the 1.0 release history, the 1.0 preview hi #### Fixed +- CLI `fetch` and `rebuild` no longer replay pending file updates from the previous database during startup. The saved queue is discarded only after the local database reset succeeds. + - CLI `fetch` and `daemon` can write received files to the Vault when `handleFilenameCaseSensitive` is enabled. The CLI now detects whether the Vault filesystem distinguishes letter case. (#1319) - Commonlib warnings now appear on stderr without `--debug`, so deliberately skipped Journals and other important notices remain visible during ordinary CLI use.