Test SyncGuard catch-up protection in Obsidian

test/e2e-obsidian/scripts/sync-guard.ts

@@ -0,0 +1,498 @@
+import { readFile } from "node:fs/promises";
+import { join } from "node:path";
+import { evalObsidianJson } from "../runner/cli.ts";
+import {
+    assertCouchDbReachable,
+    createCouchDbDatabase,
+    deleteCouchDbDatabase,
+    loadCouchDbConfig,
+    makeUniqueDatabaseName,
+    waitForCouchDbDocs,
+    type CouchDbConfig,
+} from "../runner/couchdb.ts";
+import { discoverObsidianCli, requireObsidianBinary } from "../runner/environment.ts";
+import {
+    assertEqual,
+    createE2eCouchDbPluginData,
+    createE2eObsidianDeviceLocalState,
+    prepareRemote,
+    pushLocalChanges,
+    waitForLiveSyncCoreReady,
+    waitForLocalDatabaseEntry,
+    type LocalDatabaseEntry,
+} from "../runner/liveSyncWorkflow.ts";
+import { startObsidianLiveSyncSession, type ObsidianLiveSyncSession } from "../runner/session.ts";
+import { captureObsidianPage, withObsidianPage } from "../runner/ui.ts";
+import { createTemporaryVault, type TemporaryVault } from "../runner/vault.ts";
+
+const observerKey = "__livesyncE2eSyncGuard";
+const minimumFiles = 3;
+const notes = Array.from({ length: 6 }, (_, index) => ({
+    path: `E2E/sync-guard/incoming-${index}.md`,
+    content: [
+        `# Incoming SyncGuard note ${index}`,
+        "",
+        ...Array.from(
+            { length: 6 },
+            (_, paragraph) =>
+                `Paragraph ${paragraph} of note ${index} travels through real CouchDB Metadata and Chunks.`
+        ),
+        "",
+    ].join("\n"),
+}));
+const workspacePath = "local-workspace.md";
+const workspaceContent = "# Local workspace\n\nLocal work can continue after bypass.\n";
+const workspaceEdit = "Written after bypass.\n";
+
+type GuardObservation = {
+    phase: string;
+    observedPhases: string[];
+    finiteActivity: number;
+    forwardedIds: string[];
+    transferStoppedAt: number | null;
+    reflectionHeld: boolean;
+    storageApplying: number;
+    parsing: number;
+    reflecting: number;
+    startError: string | null;
+    visible: boolean;
+    stage: string;
+    count: string;
+};
+
+async function readObservation(cliBinary: string, env: NodeJS.ProcessEnv): Promise<GuardObservation> {
+    return await evalObsidianJson<GuardObservation>(
+        cliBinary,
+        [
+            "(()=>{",
+            `const state=globalThis[${JSON.stringify(observerKey)}];`,
+            "const core=app.plugins.plugins['obsidian-livesync'].core;",
+            "const overlay=document.querySelector('.sls-sync-guard');",
+            "return JSON.stringify({",
+            "phase:core.services.replicator.replicationStatics.value.couchDB?.phase??'idle',",
+            "observedPhases:state.observedPhases,finiteActivity:core.services.replicator.finiteReplicationActivityCount.value,",
+            "forwardedIds:[...state.forwardedIds],transferStoppedAt:state.transferStoppedAt,",
+            "reflectionHeld:state.reflectionHeld,storageApplying:core.services.replication.storageApplyingCount.value,",
+            "parsing:state.parsing,reflecting:state.reflecting,startError:state.startError,",
+            "visible:!!overlay&&overlay.getBoundingClientRect().width>0,",
+            "stage:overlay?.querySelector('.sls-sync-guard-stage')?.textContent??'',",
+            "count:overlay?.querySelector('.sls-sync-guard-count')?.textContent??'',",
+            "});})()",
+        ].join(""),
+        env
+    );
+}
+
+function assertCatchUpPending(state: GuardObservation): void {
+    assertEqual(state.observedPhases.includes("catch-up"), true, "The provider never entered CouchDB catch-up.");
+    assertEqual(
+        state.observedPhases.includes("continuous"),
+        false,
+        "Catch-up completed before the held receive callback."
+    );
+    assertEqual(state.finiteActivity > 0, true, "The held receive callback lost its finite replication activity.");
+    assertEqual(state.parsing > 0, true, "The real receive callback was no longer pending.");
+    assertEqual(
+        state.phase === "catch-up" || state.phase === "idle",
+        true,
+        `Unexpected phase while the finite catch-up still owns its receive callback: ${state.phase}`
+    );
+}
+
+async function waitForObservation(
+    cliBinary: string,
+    env: NodeJS.ProcessEnv,
+    description: string,
+    predicate: (state: GuardObservation) => boolean
+): Promise<GuardObservation> {
+    const deadline = Date.now() + Number(process.env.E2E_OBSIDIAN_REMOTE_ACTIVITY_TIMEOUT_MS ?? 30000);
+    let state: GuardObservation;
+    do {
+        state = await readObservation(cliBinary, env);
+        if (state.startError) throw new Error(`Continuous replication failed: ${state.startError}`);
+        if (predicate(state)) return state;
+        await new Promise((resolve) => setTimeout(resolve, 100));
+    } while (Date.now() < deadline);
+    throw new Error(`Timed out waiting for ${description}: ${JSON.stringify(state)}`);
+}
+
+async function installGates(
+    cliBinary: string,
+    env: NodeJS.ProcessEnv,
+    incomingEntries: LocalDatabaseEntry[]
+): Promise<void> {
+    await evalObsidianJson(
+        cliBinary,
+        [
+            "(()=>{",
+            `const key=${JSON.stringify(observerKey)};`,
+            "if(globalThis[key]) throw new Error('SyncGuard gates are already installed');",
+            "const services=app.plugins.plugins['obsidian-livesync'].core.services;",
+            "const replication=services.replication;",
+            `const incomingIds=new Set(${JSON.stringify(incomingEntries.map((entry) => entry.id))});`,
+            `const heldPath=${JSON.stringify(notes[0].path)};`,
+            `const minimumFiles=${minimumFiles};`,
+            "const state={forwardedIds:new Set(),transferStoppedAt:null,reflectionHeld:false,",
+            "parsing:0,reflecting:0,startError:null,releases:[],restore:null,observedPhases:[]};",
+            "const observePhase=()=>{const phase=services.replicator.replicationStatics.value.couchDB?.phase??'idle';",
+            "if(state.observedPhases.at(-1)!==phase) state.observedPhases.push(phase);};",
+            "services.replicator.replicationStatics.onChanged(observePhase);observePhase();",
+            "const gate=()=>{let release;const promise=new Promise((resolve)=>{release=resolve;});",
+            "state.releases.push(release);return {promise,release};};",
+            "const below=gate(),threshold=gate(),reflection=gate();",
+            "state.releaseBelow=below.release;state.releaseThreshold=threshold.release;state.releaseReflection=reflection.release;",
+            "const originalParse=replication.parseSynchroniseResult;",
+            "const originalReflect=replication.processSynchroniseResult;",
+            // Forward only documents supplied by the real pull. Pausing its awaited
+            // callback retains the provider's own catch-up lifetime and counters.
+            "const parse=Object.assign(async(documents)=>{",
+            "state.parsing++;try{let accepted=true;for(const document of documents){",
+            "accepted=(await originalParse([document]))&&accepted;",
+            "if(!incomingIds.has(document._id)||state.forwardedIds.has(document._id)) continue;",
+            "state.forwardedIds.add(document._id);",
+            "if(state.forwardedIds.size===minimumFiles-1){state.transferStoppedAt=minimumFiles-1;await below.promise;state.transferStoppedAt=null;}",
+            "if(state.forwardedIds.size===minimumFiles){state.transferStoppedAt=minimumFiles;await threshold.promise;state.transferStoppedAt=null;}",
+            "}return accepted;}finally{state.parsing--;}",
+            "},originalParse);",
+            // Result processing has already incremented storageApplyingCount when
+            // this real reflection enters. The test never assigns queue counts.
+            "const reflect=Object.assign(async(entry)=>{",
+            "state.reflecting++;try{if(entry.path===heldPath){state.reflectionHeld=true;await reflection.promise;state.reflectionHeld=false;}",
+            "return await originalReflect(entry);}finally{state.reflecting--;}",
+            "},originalReflect);",
+            "replication.parseSynchroniseResult=parse;replication.processSynchroniseResult=reflect;",
+            "state.restore=()=>{",
+            "services.replicator.replicationStatics.offChanged(observePhase);",
+            "if(replication.parseSynchroniseResult===parse) replication.parseSynchroniseResult=originalParse;",
+            "if(replication.processSynchroniseResult===reflect) replication.processSynchroniseResult=originalReflect;",
+            "for(const release of state.releases) release();};",
+            "globalThis[key]=state;return JSON.stringify(true);})()",
+        ].join(""),
+        env
+    );
+}
+
+async function releaseGate(
+    cliBinary: string,
+    env: NodeJS.ProcessEnv,
+    gate: "Below" | "Threshold" | "Reflection"
+): Promise<void> {
+    await evalObsidianJson(
+        cliBinary,
+        `(()=>{globalThis[${JSON.stringify(observerKey)}].release${gate}();return JSON.stringify(true);})()`,
+        env
+    );
+}
+
+async function removeGates(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson(
+        cliBinary,
+        `(()=>{globalThis[${JSON.stringify(observerKey)}]?.restore();return JSON.stringify(true);})()`,
+        env
+    );
+    await waitForObservation(
+        cliBinary,
+        env,
+        "released receive and reflection hooks",
+        (state) => state.parsing === 0 && state.reflecting === 0
+    );
+    await evalObsidianJson(
+        cliBinary,
+        `(()=>{delete globalThis[${JSON.stringify(observerKey)}];return JSON.stringify(true);})()`,
+        env
+    );
+}
+
+async function startCatchUp(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson(
+        cliBinary,
+        [
+            "(async()=>{",
+            "const core=app.plugins.plugins['obsidian-livesync'].core;",
+            "await core.services.setting.setSyncGuardSetting('syncGuardEnabled',true);",
+            `await core.services.setting.setSyncGuardSetting('syncGuardMinimumFiles',${minimumFiles});`,
+            `const state=globalThis[${JSON.stringify(observerKey)}];`,
+            "state.startTask=(async()=>{",
+            "await core.services.setting.applyExternalSettings({liveSync:true},true);",
+            "await core.services.control.applySettings();",
+            "})().catch((error)=>{state.startError=String(error);});",
+            "return JSON.stringify(core.services.setting.syncGuardSettings);})()",
+        ].join(""),
+        env
+    );
+}
+
+async function assertVaultPathMissing(vault: TemporaryVault, path: string): Promise<void> {
+    try {
+        await readFile(join(vault.path, path), "utf8");
+    } catch (error) {
+        if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
+        throw error;
+    }
+    throw new Error(`The held reflection wrote its Vault note before release: ${path}`);
+}
+
+async function waitForVaultContent(vault: TemporaryVault, path: string, expected: string): Promise<void> {
+    const deadline = Date.now() + Number(process.env.E2E_OBSIDIAN_FILE_TIMEOUT_MS ?? 15000);
+    let actual: string | null = null;
+    do {
+        try {
+            actual = await readFile(join(vault.path, path), "utf8");
+            if (actual === expected) return;
+        } catch (error) {
+            if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
+        }
+        await new Promise((resolve) => setTimeout(resolve, 100));
+    } while (Date.now() < deadline);
+    throw new Error(`Vault content did not converge for ${path}: ${JSON.stringify(actual)}`);
+}
+
+async function assertIncomingContents(cliBinary: string, env: NodeJS.ProcessEnv, vault: TemporaryVault): Promise<void> {
+    for (const note of notes) await waitForVaultContent(vault, note.path, note.content);
+    const contents = await evalObsidianJson<string[]>(
+        cliBinary,
+        `(async()=>{const paths=${JSON.stringify(notes.map((note) => note.path))};return JSON.stringify(await Promise.all(paths.map(async(path)=>{const file=app.vault.getAbstractFileByPath(path);return file?await app.vault.read(file):null;})));})()`,
+        env
+    );
+    notes.forEach((note, index) =>
+        assertEqual(contents[index], note.content, `Obsidian read back incorrect content at ${note.path}.`)
+    );
+}
+
+async function createWorkspaceNote(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson(
+        cliBinary,
+        `(async()=>{const file=await app.vault.create(${JSON.stringify(workspacePath)},${JSON.stringify(workspaceContent)});const leaf=app.workspace.getLeaf(false);await leaf.openFile(file,{state:{mode:'source'}});leaf.view.editor.focus();return JSON.stringify(true);})()`,
+        env
+    );
+}
+
+async function runTarget(
+    binary: string,
+    cliBinary: string,
+    couchDb: CouchDbConfig,
+    dbName: string,
+    incomingEntries: LocalDatabaseEntry[],
+    bypass: boolean
+): Promise<void> {
+    const label = bypass ? "bypass" : "automatic";
+    const vault = await createTemporaryVault(`obsidian-livesync-sync-guard-${label}-`);
+    let session: ObsidianLiveSyncSession | undefined;
+    try {
+        session = await startObsidianLiveSyncSession({
+            binary,
+            cliBinary,
+            vault,
+            pluginData: createE2eCouchDbPluginData({ ...couchDb, dbName }),
+            localStorageEntries: createE2eObsidianDeviceLocalState(vault.name),
+        });
+        const { cliEnv } = session;
+        await waitForLiveSyncCoreReady(cliBinary, cliEnv);
+        await prepareRemote(cliBinary, cliEnv);
+        if (bypass) await createWorkspaceNote(cliBinary, cliEnv);
+        await installGates(cliBinary, cliEnv, incomingEntries);
+        await startCatchUp(cliBinary, cliEnv);
+
+        const below = await waitForObservation(
+            cliBinary,
+            cliEnv,
+            "real incoming notes below the threshold",
+            (state) => state.transferStoppedAt === minimumFiles - 1
+        );
+        assertCatchUpPending(below);
+        assertEqual(below.forwardedIds.length, minimumFiles - 1, "Unexpected incoming note count below the threshold.");
+        assertEqual(below.visible, false, "SyncGuard blocked the workspace before the configured threshold.");
+        await releaseGate(cliBinary, cliEnv, "Below");
+
+        const threshold = await waitForObservation(
+            cliBinary,
+            cliEnv,
+            "SyncGuard at the configured threshold",
+            (state) => state.transferStoppedAt === minimumFiles && state.visible
+        );
+        assertCatchUpPending(threshold);
+        assertEqual(
+            threshold.forwardedIds.length,
+            minimumFiles,
+            "SyncGuard did not activate at the configured threshold."
+        );
+        assertEqual(
+            threshold.count,
+            `At least ${minimumFiles} changed files received`,
+            "SyncGuard showed an incorrect received-file count."
+        );
+        assertEqual(threshold.stage, "Receiving changes from the server…", "SyncGuard did not show transfer progress.");
+        console.log(
+            await captureObsidianPage(
+                session.remoteDebuggingPort,
+                `sync-guard-${label}-threshold.png`,
+                async (page) => {
+                    await page
+                        .getByRole("button", { name: "Continue without waiting", exact: true })
+                        .waitFor({ state: "visible" });
+                }
+            )
+        );
+
+        if (bypass) {
+            const editorFocused = await evalObsidianJson<boolean>(
+                cliBinary,
+                "(()=>{const editor=app.workspace.activeLeaf.view.editor;editor.focus();return JSON.stringify(editor.hasFocus());})()",
+                cliEnv
+            );
+            assertEqual(editorFocused, true, "The guarded keyboard check did not focus the editor.");
+            await withObsidianPage(session.remoteDebuggingPort, async (page) => {
+                await page.keyboard.type("This input must stay blocked.");
+            });
+            const guardedContent = await evalObsidianJson<string>(
+                cliBinary,
+                "(()=>JSON.stringify(app.workspace.activeLeaf.view.editor.getValue()))()",
+                cliEnv
+            );
+            assertEqual(guardedContent, workspaceContent, "The protected editor accepted keyboard input.");
+            await withObsidianPage(session.remoteDebuggingPort, async (page) => {
+                await page.getByRole("button", { name: "Continue without waiting", exact: true }).click();
+                await page.locator(".sls-sync-guard").waitFor({ state: "hidden" });
+                const editor = page.locator(".workspace-leaf.mod-active .cm-content");
+                await editor.click();
+                await page.keyboard.press("ControlOrMeta+End");
+                await page.keyboard.type(workspaceEdit);
+            });
+            await waitForVaultContent(vault, workspacePath, workspaceContent + workspaceEdit);
+            const continuing = await readObservation(cliBinary, cliEnv);
+            assertCatchUpPending(continuing);
+            assertEqual(
+                continuing.transferStoppedAt,
+                minimumFiles,
+                "The receive callback was not pending during bypass."
+            );
+            assertEqual(continuing.visible, false, "The bypass did not restore the workspace.");
+            await releaseGate(cliBinary, cliEnv, "Threshold");
+            await waitForObservation(
+                cliBinary,
+                cliEnv,
+                "continuous replication after bypass",
+                (state) => state.phase === "continuous" && state.reflectionHeld
+            );
+            await assertVaultPathMissing(vault, notes[0].path);
+            assertEqual(
+                (await readObservation(cliBinary, cliEnv)).visible,
+                false,
+                "The guard reopened during the bypassed catch-up."
+            );
+        } else {
+            await releaseGate(cliBinary, cliEnv, "Threshold");
+            const applying = await waitForObservation(
+                cliBinary,
+                cliEnv,
+                "network completion with a held Vault reflection",
+                (state) => state.phase === "continuous" && state.reflectionHeld && state.storageApplying > 0
+            );
+            assertEqual(
+                applying.forwardedIds.length,
+                notes.length,
+                "The real catch-up did not deliver every incoming note."
+            );
+            assertEqual(applying.visible, true, "SyncGuard released before the held Vault reflection finished.");
+            assertEqual(
+                applying.stage,
+                "Applying received changes to local notes…",
+                "SyncGuard did not show local application progress."
+            );
+            await assertVaultPathMissing(vault, notes[0].path);
+            // Exceed the guard's ordinary quiet period while real application work remains pending.
+            await new Promise((resolve) => setTimeout(resolve, 2500));
+            const stillApplying = await readObservation(cliBinary, cliEnv);
+            assertEqual(stillApplying.visible, true, "SyncGuard released while real reflection was still pending.");
+            assertEqual(stillApplying.reflectionHeld, true, "The reflection gate was released unexpectedly.");
+            console.log(
+                await captureObsidianPage(session.remoteDebuggingPort, "sync-guard-applying.png", async (page) => {
+                    await page.locator(".sls-sync-guard").waitFor({ state: "visible" });
+                })
+            );
+        }
+
+        await releaseGate(cliBinary, cliEnv, "Reflection");
+        await assertIncomingContents(cliBinary, cliEnv, vault);
+        await waitForObservation(
+            cliBinary,
+            cliEnv,
+            "workspace release after exact Vault reflection",
+            (state) => !state.visible && state.phase === "continuous" && state.storageApplying === 0
+        );
+        if (bypass) await waitForVaultContent(vault, workspacePath, workspaceContent + workspaceEdit);
+        console.log(
+            `SyncGuard ${label}: threshold ${minimumFiles}, ${notes.length} exact incoming notes, workspace released.`
+        );
+    } catch (error) {
+        if (session) {
+            await captureObsidianPage(
+                session.remoteDebuggingPort,
+                `sync-guard-${label}-failure.png`,
+                async () => {}
+            ).catch(() => undefined);
+            const observation = await readObservation(cliBinary, session.cliEnv).catch(() => null);
+            console.error(`SyncGuard ${label} failure state: ${JSON.stringify(observation)}`);
+        }
+        throw error;
+    } finally {
+        if (session) {
+            await removeGates(cliBinary, session.cliEnv).catch((error: unknown) => {
+                console.warn(`SyncGuard hook cleanup: ${error instanceof Error ? error.message : String(error)}`);
+            });
+            await session.app.stop();
+        }
+        await vault.dispose();
+    }
+}
+
+async function main(): Promise<void> {
+    const binary = requireObsidianBinary();
+    const cli = discoverObsidianCli();
+    if (!cli.binary) throw new Error(`Could not find obsidian-cli. Checked paths: ${cli.checked.join(", ")}`);
+    const couchDb = await loadCouchDbConfig();
+    const dbName = makeUniqueDatabaseName(couchDb.dbPrefix, "sync-guard");
+    const sourceVault = await createTemporaryVault("obsidian-livesync-sync-guard-source-");
+    let source: ObsidianLiveSyncSession | undefined;
+    try {
+        await assertCouchDbReachable(couchDb);
+        await createCouchDbDatabase(couchDb, dbName);
+        source = await startObsidianLiveSyncSession({
+            binary,
+            cliBinary: cli.binary,
+            vault: sourceVault,
+            pluginData: createE2eCouchDbPluginData({ ...couchDb, dbName }),
+            localStorageEntries: createE2eObsidianDeviceLocalState(sourceVault.name),
+        });
+        await waitForLiveSyncCoreReady(cli.binary, source.cliEnv);
+        await prepareRemote(cli.binary, source.cliEnv);
+        await evalObsidianJson(
+            cli.binary,
+            `(async()=>{const notes=${JSON.stringify(notes)};await app.vault.createFolder('E2E');await app.vault.createFolder('E2E/sync-guard');for(const note of notes){const file=await app.vault.create(note.path,note.content);if(await app.vault.read(file)!==note.content) throw new Error('Source note read-back failed');}return JSON.stringify(true);})()`,
+            source.cliEnv
+        );
+        const entries: LocalDatabaseEntry[] = [];
+        for (const note of notes) entries.push(await waitForLocalDatabaseEntry(cli.binary, source.cliEnv, note.path));
+        if (entries.some((entry) => entry.children.length === 0))
+            throw new Error("Source notes did not produce real Chunks.");
+        await pushLocalChanges(cli.binary, source.cliEnv);
+        await waitForCouchDbDocs(couchDb, dbName, (documents) => {
+            const ids = new Set(documents.map((document) => document._id));
+            return entries.every((entry) => ids.has(entry.id) && entry.children.every((id) => ids.has(id)));
+        });
+        await source.app.stop();
+        source = undefined;
+
+        await runTarget(binary, cli.binary, couchDb, dbName, entries, false);
+        await runTarget(binary, cli.binary, couchDb, dbName, entries, true);
+    } finally {
+        if (source) await source.app.stop();
+        await sourceVault.dispose();
+        if (process.env.E2E_OBSIDIAN_KEEP_COUCHDB !== "true") await deleteCouchDbDatabase(couchDb, dbName);
+    }
+}
+
+main().catch((error: unknown) => {
+    console.error(error instanceof Error ? error.stack : error);
+    process.exit(1);
+});