Prevent old reflection replay during CLI reconstruction

src/LiveSyncBaseCore.ts

@@ -40,6 +40,7 @@ export interface LiveSyncCoreFeatureViews {
 }
 
 export interface StartupDatabaseOptions {
+    readonly discardReplicationResultSnapshotOnReset?: boolean;
     readonly ignoreSuspending?: boolean;
     readonly continueOnFileFailure?: boolean;
 }

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<CliCommandRunner>(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<CliCommandRunner>(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 },

src/apps/cli/main.ts

@@ -77,13 +77,21 @@ const VAULT_SYNC_MODES: Readonly<Partial<Record<CLICommand, CLIVaultSyncMode>>>
         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,
+        },
     },
 };
 

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<void> {
+        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<ReplicateResultProcessorState>(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<EntryDoc>) {
+        if (this.waitingForDatabaseReset) return;
         const old = this._queuedChanges.find((e) => e._id == doc._id);
         const path = "path" in doc ? this.getPath(doc) : "<unknown>";
         const docNote = `${path} (${shortenId(doc._id)}, ${shortenRev(doc._rev)})`;

src/serviceFeatures/replication/ReplicateResultProcessor.unit.spec.ts

@@ -37,6 +37,7 @@ function note(id: string): PouchDB.Core.ExistingDocument<EntryDoc> {
 }
 
 type SetupOptions = {
+    waitForDatabaseReset?: boolean;
     onVaultReflectionFailure?: () => void;
     getRaw?: (id: string) => Promise<unknown>;
     getDBEntryFromMeta?: (entry: object) => Promise<unknown>;
@@ -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({

src/serviceFeatures/replication/index.ts

@@ -37,28 +37,38 @@ export function useReplicationFeature<TContext extends ServiceContext, TCommands
     const localApplicationActivityOwner = ownsLocalApplicationActivity(services.replicator)
         ? services.replicator
         : undefined;
-    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());
+    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(),