fix: align chunk reads with remote delivery

Consume commonlib's delivery lifecycle, classify finite replication entry points, and document why the five-minute inactivity fuse is only a leak safety valve. Record the remote-activity counter correction and preserve continuous replication's unbounded live channel.
2026-07-17 10:06:00 +00:00 · 2026-07-15 17:39:39 +00:00
parent f54d162ef9
commit 9f9758bc12
16 changed files with 792 additions and 28 deletions
@@ -126,6 +126,26 @@ export async function createCouchDbDatabase(config: CouchDbConfig, dbName: strin
    }
 }

+export async function putCouchDbDocument(
+    config: CouchDbConfig,
+    dbName: string,
+    document: CouchDbDocument
+): Promise<void> {
+    const response = await fetch(databaseUrl(config, dbName, `/${encodeURIComponent(document._id)}`), {
+        method: "PUT",
+        headers: {
+            authorization: authHeader(config),
+            "content-type": "application/json",
+        },
+        body: JSON.stringify(document),
+    });
+    if (!response.ok) {
+        throw new Error(
+            `Failed to write CouchDB document ${document._id}. HTTP ${response.status}: ${await response.text()}`
+        );
+    }
+}
+
 export async function deleteCouchDbDatabase(config: CouchDbConfig, dbName: string): Promise<void> {
    const response = await fetch(databaseUrl(config, dbName), {
        method: "DELETE",
@@ -0,0 +1,180 @@
+import { mkdir, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+import type { Page } from "playwright";
+import { withObsidianPage } from "./ui.ts";
+
+export const REMOTE_ACTIVITY_E2E_STATE_KEY = "__livesyncE2ERemoteActivity";
+
+export type RemoteActivitySnapshot = {
+    boundedRemoteActivityCount: number;
+    finiteReplicationActivityCount: number;
+    gateDone?: boolean;
+    gateEntered?: boolean;
+    gateError?: string;
+    gateKind?: string;
+    requestCount: number;
+    responseCount: number;
+    statusBarFound: boolean;
+    statusBarText: string;
+};
+
+export type ExpectedRemoteActivityState = "chunk-fetch-active" | "finite-replication-active" | "idle";
+
+type RuntimeCounter = { value?: number };
+
+type RuntimeCore = {
+    services?: {
+        API?: {
+            requestCount?: RuntimeCounter;
+            responseCount?: RuntimeCounter;
+        };
+        replicator?: {
+            boundedRemoteActivityCount?: RuntimeCounter;
+            finiteReplicationActivityCount?: RuntimeCounter;
+        };
+    };
+};
+
+type RuntimeGate = {
+    done?: boolean;
+    entered?: boolean;
+    error?: string;
+    kind?: string;
+};
+
+type RendererGlobals = typeof globalThis & {
+    app?: {
+        plugins?: {
+            plugins?: Record<string, { core?: RuntimeCore }>;
+        };
+    };
+};
+
+async function readRemoteActivitySnapshotFromPage(page: Page): Promise<RemoteActivitySnapshot> {
+    return await page.evaluate(
+        ({ pluginId, stateKey }) => {
+            const globals = globalThis as RendererGlobals;
+            const core = globals.app?.plugins?.plugins?.[pluginId]?.core;
+            if (!core) throw new Error(`Obsidian plug-in is not loaded: ${pluginId}`);
+            const gate = (globalThis as unknown as Record<string, RuntimeGate | undefined>)[stateKey];
+            const statusBars = Array.from(document.querySelectorAll<HTMLElement>(".syncstatusbar"));
+            return {
+                boundedRemoteActivityCount: Number(core.services?.replicator?.boundedRemoteActivityCount?.value ?? -1),
+                finiteReplicationActivityCount: Number(
+                    core.services?.replicator?.finiteReplicationActivityCount?.value ?? -1
+                ),
+                gateDone: gate?.done,
+                gateEntered: gate?.entered,
+                gateError: gate?.error,
+                gateKind: gate?.kind,
+                requestCount: Number(core.services?.API?.requestCount?.value ?? -1),
+                responseCount: Number(core.services?.API?.responseCount?.value ?? -1),
+                statusBarFound: statusBars.length > 0,
+                statusBarText: statusBars.map((element) => element.textContent ?? "").join("\n"),
+            } satisfies RemoteActivitySnapshot;
+        },
+        { pluginId: "obsidian-livesync", stateKey: REMOTE_ACTIVITY_E2E_STATE_KEY }
+    );
+}
+
+export async function readRemoteActivitySnapshot(port: number): Promise<RemoteActivitySnapshot> {
+    return await withObsidianPage(port, async (page) => await readRemoteActivitySnapshotFromPage(page));
+}
+
+function formatWaitFailure(
+    expected: ExpectedRemoteActivityState,
+    snapshot: RemoteActivitySnapshot | undefined,
+    error: unknown
+): Error {
+    return new Error(
+        [
+            `Timed out waiting for remote activity state: ${expected}`,
+            snapshot ? `Last snapshot: ${JSON.stringify(snapshot)}` : undefined,
+            error instanceof Error ? error.message : String(error),
+        ]
+            .filter((line): line is string => line !== undefined)
+            .join("\n")
+    );
+}
+
+export async function waitForRemoteActivityState(
+    port: number,
+    expected: ExpectedRemoteActivityState,
+    timeoutMs = Number(process.env.E2E_OBSIDIAN_REMOTE_ACTIVITY_TIMEOUT_MS ?? 30000)
+): Promise<RemoteActivitySnapshot> {
+    try {
+        return await withObsidianPage(port, async (page) => {
+            await page.waitForFunction(
+                ({ expectedState, pluginId, stateKey }) => {
+                    const globals = globalThis as RendererGlobals;
+                    const core = globals.app?.plugins?.plugins?.[pluginId]?.core;
+                    if (!core) return false;
+                    const gate = (globalThis as unknown as Record<string, RuntimeGate | undefined>)[stateKey];
+                    const statusBarText = Array.from(document.querySelectorAll<HTMLElement>(".syncstatusbar"))
+                        .map((element) => element.textContent ?? "")
+                        .join("\n");
+                    const bounded = Number(core.services?.replicator?.boundedRemoteActivityCount?.value ?? -1);
+                    const finite = Number(core.services?.replicator?.finiteReplicationActivityCount?.value ?? -1);
+                    const requests = Number(core.services?.API?.requestCount?.value ?? -1);
+                    const responses = Number(core.services?.API?.responseCount?.value ?? -1);
+                    const iconVisible = statusBarText.includes("📲");
+
+                    if (expectedState === "finite-replication-active") {
+                        return (
+                            gate?.kind === "one-shot" &&
+                            gate.entered === true &&
+                            bounded > 0 &&
+                            finite > 0 &&
+                            iconVisible
+                        );
+                    }
+                    if (expectedState === "chunk-fetch-active") {
+                        return (
+                            gate?.kind === "chunk-fetch" &&
+                            gate.entered === true &&
+                            bounded > 0 &&
+                            finite === 0 &&
+                            iconVisible
+                        );
+                    }
+                    return bounded === 0 && finite === 0 && requests === responses && !iconVisible;
+                },
+                {
+                    expectedState: expected,
+                    pluginId: "obsidian-livesync",
+                    stateKey: REMOTE_ACTIVITY_E2E_STATE_KEY,
+                },
+                { timeout: timeoutMs }
+            );
+            return await readRemoteActivitySnapshotFromPage(page);
+        });
+    } catch (error) {
+        const snapshot = await readRemoteActivitySnapshot(port).catch(() => undefined);
+        throw formatWaitFailure(expected, snapshot, error);
+    }
+}
+
+export type RemoteActivityDiagnostics = {
+    screenshotPath: string;
+    snapshot: RemoteActivitySnapshot;
+    snapshotPath: string;
+};
+
+export async function captureRemoteActivityDiagnostics(
+    port: number,
+    label: string
+): Promise<RemoteActivityDiagnostics> {
+    const outputDirectory = process.env.E2E_OBSIDIAN_DIAGNOSTICS_DIR ?? "/tmp/obsidian-livesync-e2e";
+    await mkdir(outputDirectory, { recursive: true });
+    const safeLabel = label.replace(/[^a-z0-9_-]+/gi, "-").replace(/^-+|-+$/g, "") || "remote-activity";
+    const prefix = `${safeLabel}-${new Date().toISOString().replace(/[:.]/g, "-")}`;
+    const screenshotPath = join(outputDirectory, `${prefix}.png`);
+    const snapshotPath = join(outputDirectory, `${prefix}.json`);
+    const snapshot = await withObsidianPage(port, async (page) => {
+        const current = await readRemoteActivitySnapshotFromPage(page);
+        await page.screenshot({ path: screenshotPath, fullPage: true });
+        return current;
+    });
+    await writeFile(snapshotPath, `${JSON.stringify(snapshot, undefined, 2)}\n`, "utf8");
+    return { screenshotPath, snapshot, snapshotPath };
+}
@@ -0,0 +1,187 @@
+import { evalObsidianJson } from "./cli.ts";
+import { REMOTE_ACTIVITY_E2E_STATE_KEY } from "./remoteActivity.ts";
+
+export type HeldRemoteActivityResult = {
+    done: boolean;
+    entered: boolean;
+    error?: string;
+    kind: "chunk-fetch" | "one-shot";
+    requestedIds?: string[];
+    result?: boolean;
+    resultCount?: number;
+};
+
+const stateKeySource = JSON.stringify(REMOTE_ACTIVITY_E2E_STATE_KEY);
+
+export async function startHeldOneShotReplication(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson<{ started: boolean }>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const stateKey=${stateKeySource};`,
+            "const host=globalThis;",
+            "if(host[stateKey]) throw new Error('A remote activity E2E gate is already installed.');",
+            "const core=app.plugins.plugins['obsidian-livesync'].core;",
+            "const replicator=core.services.replicator.getActiveReplicator();",
+            "if(!replicator) throw new Error('No active replicator is available.');",
+            "const original=replicator.openReplication;",
+            "let releaseGate;",
+            "const gate=new Promise((resolve)=>{releaseGate=resolve;});",
+            "const state={kind:'one-shot',entered:false,done:false,released:false,error:undefined,result:undefined,promise:undefined,release:undefined,restore:undefined};",
+            "state.release=()=>{if(!state.released){state.released=true;releaseGate();}};",
+            "state.restore=()=>{replicator.openReplication=original;};",
+            "host[stateKey]=state;",
+            "replicator.openReplication=async function(...args){",
+            "state.entered=true;",
+            "await gate;",
+            "return await original.apply(this,args);",
+            "};",
+            "state.promise=(async()=>{",
+            "try{",
+            "if(!(await core.services.fileProcessing.commitPendingFileEvents())) throw new Error('Pending file events could not be committed.');",
+            "state.result=!!(await core.services.replication.replicate(true));",
+            "}catch(error){",
+            "state.error=error instanceof Error?error.message:String(error);",
+            "}finally{",
+            "state.restore();",
+            "state.done=true;",
+            "}",
+            "})();",
+            "return JSON.stringify({started:true});",
+            "})()",
+        ].join(""),
+        env
+    );
+}
+
+export async function startHeldChunkFetch(cliBinary: string, env: NodeJS.ProcessEnv, chunkId: string): Promise<void> {
+    await evalObsidianJson<{ started: boolean }>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const stateKey=${stateKeySource};`,
+            `const chunkId=${JSON.stringify(chunkId)};`,
+            "const host=globalThis;",
+            "if(host[stateKey]) throw new Error('A remote activity E2E gate is already installed.');",
+            "const core=app.plugins.plugins['obsidian-livesync'].core;",
+            "const replicator=core.services.replicator.getActiveReplicator();",
+            "if(!replicator) throw new Error('No active replicator is available.');",
+            "const localDb=core.localDatabase.localDatabase;",
+            "const existing=await localDb.get(chunkId).catch(()=>undefined);",
+            "if(existing&&!existing._deleted) throw new Error(`The remote-only chunk already exists locally: ${chunkId}`);",
+            "const original=replicator.fetchRemoteChunks;",
+            "let releaseGate;",
+            "let resolveDone;",
+            "const gate=new Promise((resolve)=>{releaseGate=resolve;});",
+            "const donePromise=new Promise((resolve)=>{resolveDone=resolve;});",
+            "const state={kind:'chunk-fetch',entered:false,done:false,released:false,error:undefined,resultCount:undefined,requestedIds:undefined,promise:donePromise,release:undefined,restore:undefined};",
+            "state.release=()=>{if(!state.released){state.released=true;releaseGate();}};",
+            "state.restore=()=>{replicator.fetchRemoteChunks=original;};",
+            "host[stateKey]=state;",
+            "replicator.fetchRemoteChunks=async function(...args){",
+            "state.entered=true;",
+            "state.requestedIds=Array.isArray(args[0])?[...args[0]]:[];",
+            "await gate;",
+            "try{",
+            "const result=await original.apply(this,args);",
+            "state.resultCount=Array.isArray(result)?result.length:0;",
+            "return result;",
+            "}catch(error){",
+            "state.error=error instanceof Error?error.message:String(error);",
+            "throw error;",
+            "}finally{",
+            "state.restore();",
+            "state.done=true;",
+            "resolveDone();",
+            "}",
+            "};",
+            "core.localDatabase.managers.chunkFetcher.onEvent([chunkId]);",
+            "return JSON.stringify({started:true});",
+            "})()",
+        ].join(""),
+        env
+    );
+}
+
+export async function finishHeldRemoteActivity(
+    cliBinary: string,
+    env: NodeJS.ProcessEnv
+): Promise<HeldRemoteActivityResult> {
+    return await evalObsidianJson<HeldRemoteActivityResult>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const stateKey=${stateKeySource};`,
+            "const state=globalThis[stateKey];",
+            "if(!state) throw new Error('No remote activity E2E gate is installed.');",
+            "state.release();",
+            "await state.promise;",
+            "return JSON.stringify({kind:state.kind,entered:state.entered,done:state.done,error:state.error,result:state.result,resultCount:state.resultCount,requestedIds:state.requestedIds});",
+            "})()",
+        ].join(""),
+        env
+    );
+}
+
+export async function waitForRestoredChunk(
+    cliBinary: string,
+    env: NodeJS.ProcessEnv,
+    chunkId: string,
+    timeoutMs = Number(process.env.E2E_OBSIDIAN_REMOTE_ACTIVITY_TIMEOUT_MS ?? 30000)
+): Promise<{ id: string; type: string }> {
+    return await evalObsidianJson<{ id: string; type: string }>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const chunkId=${JSON.stringify(chunkId)};`,
+            `const deadline=Date.now()+${JSON.stringify(timeoutMs)};`,
+            "const core=app.plugins.plugins['obsidian-livesync'].core;",
+            "const localDb=core.localDatabase.localDatabase;",
+            "const sleep=(ms)=>new Promise((resolve)=>setTimeout(resolve,ms));",
+            "while(Date.now()<deadline){",
+            "const chunk=await localDb.get(chunkId).catch(()=>undefined);",
+            "if(chunk&&!chunk._deleted&&chunk.type==='leaf'&&typeof chunk.data==='string') return JSON.stringify({id:chunk._id,type:chunk.type});",
+            "await sleep(100);",
+            "}",
+            "throw new Error(`Timed out waiting for the fetched chunk to return: ${chunkId}`);",
+            "})()",
+        ].join(""),
+        env
+    );
+}
+
+export async function clearHeldRemoteActivity(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson<{ cleared: boolean }>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const stateKey=${stateKeySource};`,
+            "const state=globalThis[stateKey];",
+            "if(!state) return JSON.stringify({cleared:false});",
+            "if(!state.done) throw new Error('The remote activity E2E gate is still running.');",
+            "delete globalThis[stateKey];",
+            "return JSON.stringify({cleared:true});",
+            "})()",
+        ].join(""),
+        env
+    );
+}
+
+export async function cleanUpHeldRemoteActivity(cliBinary: string, env: NodeJS.ProcessEnv): Promise<void> {
+    await evalObsidianJson<{ cleared: boolean }>(
+        cliBinary,
+        [
+            "(async()=>{",
+            `const stateKey=${stateKeySource};`,
+            "const state=globalThis[stateKey];",
+            "if(!state) return JSON.stringify({cleared:false});",
+            "state.release?.();",
+            "await Promise.race([Promise.resolve(state.promise),new Promise((resolve)=>setTimeout(resolve,5000))]);",
+            "state.restore?.();",
+            "delete globalThis[stateKey];",
+            "return JSON.stringify({cleared:true});",
+            "})()",
+        ].join(""),
+        env
+    );
+}