Reprocess stored documents after reflection filters change

src/modules/core/ModuleReplicator.ts

@@ -6,7 +6,11 @@ import { skipIfDuplicated } from "octagonal-wheels/concurrency/lock";
 import { balanceChunkPurgedDBs } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks";
 import { purgeUnreferencedChunks } from "@vrtmrz/livesync-commonlib/compat/pouchdb/chunks";
 import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
-import { type EntryDoc, type RemoteType } from "@vrtmrz/livesync-commonlib/compat/common/types";
+import {
+    type EntryDoc,
+    type ObsidianLiveSyncSettings,
+    type RemoteType,
+} from "@vrtmrz/livesync-commonlib/compat/common/types";
 
 import { scheduleTask } from "octagonal-wheels/concurrency/task";
 import { EVENT_FILE_SAVED, EVENT_SETTING_SAVED, eventHub } from "@/common/events";
@@ -72,18 +76,52 @@ export class ModuleReplicator extends AbstractModule {
         this._unresolvedErrorManager.clearErrors();
     }
 
+    private _normalFileReflectionFilterSignature: string | undefined;
+
+    private getNormalFileReflectionFilterSignature(
+        settings: Pick<
+            ObsidianLiveSyncSettings,
+            | "handleFilenameCaseSensitive"
+            | "ignoreFiles"
+            | "maxMTimeForReflectEvents"
+            | "syncIgnoreRegEx"
+            | "syncInternalFiles"
+            | "syncMaxSizeInMB"
+            | "syncOnlyRegEx"
+            | "useIgnoreFiles"
+        >
+    ): string {
+        return JSON.stringify({
+            handleFilenameCaseSensitive: settings.handleFilenameCaseSensitive ?? false,
+            ignoreFiles: settings.ignoreFiles ?? "",
+            maxMTimeForReflectEvents: settings.maxMTimeForReflectEvents ?? 0,
+            syncIgnoreRegEx: settings.syncIgnoreRegEx ?? "",
+            syncInternalFiles: settings.syncInternalFiles ?? false,
+            syncMaxSizeInMB: settings.syncMaxSizeInMB ?? 0,
+            syncOnlyRegEx: settings.syncOnlyRegEx ?? "",
+            useIgnoreFiles: settings.useIgnoreFiles ?? false,
+        });
+    }
+
     private _everyOnloadAfterLoadSettings(): Promise<boolean> {
+        this._normalFileReflectionFilterSignature = this.getNormalFileReflectionFilterSignature(this.settings);
         eventHub.onEvent(EVENT_FILE_SAVED, () => {
             if (this.settings.syncOnSave && !this.core.services.appLifecycle.isSuspended()) {
                 scheduleTask("perform-replicate-after-save", 250, () => this.services.replication.replicateByEvent());
             }
         });
         eventHub.onEvent(EVENT_SETTING_SAVED, (setting) => {
+            const previousReflectionFilter = this._normalFileReflectionFilterSignature;
+            const nextReflectionFilter = this.getNormalFileReflectionFilterSignature(setting);
+            this._normalFileReflectionFilterSignature = nextReflectionFilter;
             if (this.core.settings.suspendParseReplicationResult) {
                 this.processor.suspend();
             } else {
                 this.processor.resume();
             }
+            if (previousReflectionFilter !== undefined && previousReflectionFilter !== nextReflectionFilter) {
+                fireAndForget(() => this.processor.reprocessStoredDocuments());
+            }
         });
 
         return Promise.resolve(true);

src/modules/core/ModuleReplicator.unit.spec.ts

@@ -1,5 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import { createServiceContext } from "@vrtmrz/livesync-commonlib/context";
+import { EVENT_SETTING_SAVED, eventHub } from "@/common/events";
+import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
 
 const chunkMocks = vi.hoisted(() => ({
     purgeUnreferencedChunks: vi.fn(async (_db: unknown, countOnly: boolean) => (countOnly ? 2 : 0)),
@@ -58,6 +60,60 @@ describe("ModuleReplicator", () => {
 
         expect(ensurePBKDF2Salt).toHaveBeenCalledWith({}, false, false);
     });
+
+    it("reprocesses stored documents when the normal-file target filters change", async () => {
+        eventHub.offAll();
+        const settings = {
+            handleFilenameCaseSensitive: false,
+            ignoreFiles: ".gitignore",
+            maxMTimeForReflectEvents: 0,
+            syncOnlyRegEx: "^E2E/allowed/.*",
+            syncIgnoreRegEx: "",
+            syncInternalFiles: false,
+            syncMaxSizeInMB: 0,
+            suspendParseReplicationResult: false,
+            useIgnoreFiles: false,
+        } as ObsidianLiveSyncSettings;
+        const services = {
+            context: createServiceContext(),
+            API: {
+                addLog: vi.fn(),
+                addCommand: vi.fn(),
+                registerWindow: vi.fn(),
+                addRibbonIcon: vi.fn(),
+                registerProtocolHandler: vi.fn(),
+            },
+            appLifecycle: {
+                getUnresolvedMessages: { addHandler: vi.fn() },
+                isSuspended: vi.fn(() => false),
+            },
+        };
+        const core = {
+            _services: services,
+            services,
+            settings,
+        } as any;
+        const module = new ModuleReplicator(core);
+        const reprocessStoredDocuments = vi.fn(async () => 1);
+        Object.assign(module.processor, { reprocessStoredDocuments });
+
+        try {
+            await (module as any)._everyOnloadAfterLoadSettings();
+            eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
+            await Promise.resolve();
+            expect(reprocessStoredDocuments).not.toHaveBeenCalled();
+
+            Object.assign(settings, { syncOnlyRegEx: "" });
+            eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
+            await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledOnce());
+
+            settings.syncMaxSizeInMB = 10;
+            eventHub.emitEvent(EVENT_SETTING_SAVED, { ...settings });
+            await vi.waitFor(() => expect(reprocessStoredDocuments).toHaveBeenCalledTimes(2));
+        } finally {
+            eventHub.offAll();
+        }
+    });
 });
 
 describe("ModuleReplicator legacy cleanup", () => {

src/modules/core/ReplicateResultProcessor.ts

@@ -26,6 +26,7 @@ import { isNotFoundError } from "@vrtmrz/livesync-commonlib/compat/common/utils.
 import type PouchDB from "pouchdb-core";
 
 const KV_KEY_REPLICATION_RESULT_PROCESSOR_SNAPSHOT = "replicationResultProcessorSnapshot";
+const REPROCESS_BATCH_SIZE = 100;
 type ReplicateResultProcessorState = {
     queued: PouchDB.Core.ExistingDocument<EntryDoc>[];
     processing: PouchDB.Core.ExistingDocument<EntryDoc>[];
@@ -182,6 +183,26 @@ export class ReplicateResultProcessor {
             }
         }
     }
+
+    /**
+     * Requeues stored normal-file metadata after its reflection filters change.
+     * Replication checkpoints may already cover documents which were skipped
+     * by the previous filter, so a later ordinary sync cannot emit them again.
+     */
+    public async reprocessStoredDocuments(): Promise<number> {
+        let count = 0;
+        let batch: PouchDB.Core.ExistingDocument<EntryDoc>[] = [];
+        for await (const document of this.localDatabase.findAllNormalDocs()) {
+            batch.push(document);
+            count++;
+            if (batch.length < REPROCESS_BATCH_SIZE) continue;
+            this.enqueueAll(batch);
+            batch = [];
+        }
+        if (batch.length > 0) this.enqueueAll(batch);
+        this.log(`Requeued ${count} stored document(s) after the reflection filters changed`, LOG_LEVEL_INFO);
+        return count;
+    }
     /**
      * Process the change if it is not a document change.
      * @param change Change to process

src/modules/core/ReplicateResultProcessor.unit.spec.ts

@@ -0,0 +1,25 @@
+import { describe, expect, it, vi } from "vitest";
+import type { EntryDoc } from "@vrtmrz/livesync-commonlib/compat/common/types";
+import { ReplicateResultProcessor } from "./ReplicateResultProcessor";
+
+describe("ReplicateResultProcessor target-filter reprocessing", () => {
+    it("scans normal-file metadata without loading chunk documents and requeues it", async () => {
+        const documents = [
+            { _id: "first", _rev: "1-a", type: "plain", path: "first.md" },
+            { _id: "second", _rev: "1-b", type: "plain", path: "second.md" },
+        ] as unknown as PouchDB.Core.ExistingDocument<EntryDoc>[];
+        const findAllNormalDocs = vi.fn(async function* () {
+            yield* documents;
+        });
+        const processor = new ReplicateResultProcessor({
+            core: { localDatabase: { findAllNormalDocs } },
+        } as never);
+        const enqueueAll = vi.spyOn(processor, "enqueueAll").mockImplementation(() => undefined);
+
+        await expect(processor.reprocessStoredDocuments()).resolves.toBe(2);
+
+        expect(findAllNormalDocs).toHaveBeenCalledOnce();
+        expect(enqueueAll).toHaveBeenCalledOnce();
+        expect(enqueueAll).toHaveBeenCalledWith(documents);
+    });
+});

test/e2e-obsidian/scripts/two-vault-sync.ts

@@ -18,7 +18,6 @@ import {
     assertE2eCompatibilityReviewPending,
     configureCouchDb,
     createE2eCouchDbPluginData,
-    createE2eObsidianDeviceLocalState,
     prepareRemote,
     pushLocalChanges,
     resumeCompatibilityReview,
@@ -196,10 +195,6 @@ async function startConfiguredSession(
         vault,
         startupGraceMs: Number(process.env.E2E_OBSIDIAN_STARTUP_GRACE_MS ?? 1000),
         pluginData: createE2eCouchDbPluginData(couchDbSettings, overrides),
-        // The real profile retains this device-local acknowledgement. Seed it
-        // on later process launches because the isolated Electron runner does
-        // not currently preserve localStorage across those launches.
-        localStorageEntries: reviewAlreadyCompleted ? createE2eObsidianDeviceLocalState(vault.name) : undefined,
     });
     context.activeSessions.add(session);
     try {
@@ -517,6 +512,7 @@ async function runTargetMismatch(
         syncOnlyRegEx: "^E2E/two-vault/allowed/.*",
     });
     await syncAndApply(context, session);
+    await waitForLocalDatabaseEntry(context.cliBinary, session.cliEnv, targetMismatchPath);
     assertEqual(
         await pathExists(vaultB.path, targetMismatchPath),
         false,
@@ -525,8 +521,24 @@ async function runTargetMismatch(
     await stopTrackedSession(context, session);
 
     session = await startConfiguredSession(context, vaultB, {
-        syncOnlyRegEx: "",
+        syncOnlyRegEx: "^E2E/two-vault/allowed/.*",
     });
+    assertEqual(
+        await pathExists(vaultB.path, targetMismatchPath),
+        false,
+        "A checkpointed non-target note was reflected before its target filter changed."
+    );
+    await configureCouchDb(
+        context.cliBinary,
+        session.cliEnv,
+        {
+            uri: context.couchDb.uri,
+            username: context.couchDb.username,
+            password: context.couchDb.password,
+            dbName: context.dbName,
+        },
+        { syncOnlyRegEx: "" }
+    );
     await syncAndApply(context, session);
     const reflectedAfterEnabling = await waitForPathContent(
         vaultB.path,