Extract the Journal skip command and clarify manual recovery

src/apps/cli/commands/journalSkip.ts

@@ -0,0 +1,74 @@
+import type { StandardIo } from "@vrtmrz/livesync-commonlib/context";
+import {
+    isReplicationCompleted,
+    REPLICATION_PROGRESS_PRESENTATIONS,
+    USER_INITIATED_REPLICATION_AUTHORITY,
+} from "@vrtmrz/livesync-commonlib/replication";
+import { isLiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
+import { writeStderrLine, writeStdoutLine } from "@/apps/cli/cliOutput";
+import type { CLICommandContext, CLIOptions } from "./types";
+
+type JournalSkipOptions = Pick<CLIOptions, "commandArgs">;
+
+type JournalSkipContext = {
+    standardIo: StandardIo;
+    getActiveReplicator: CLICommandContext["core"]["services"]["replicator"]["getActiveReplicator"];
+    replicateUserInitiated: CLICommandContext["core"]["services"]["replication"]["replicateUserInitiated"];
+    currentSettings: CLICommandContext["core"]["services"]["setting"]["currentSettings"];
+};
+
+/** Retry replication, then skip only the exact Journal snapshot left by the same replicator. */
+export async function runJournalSkipCommand(
+    options: JournalSkipOptions,
+    context: JournalSkipContext
+): Promise<boolean> {
+    const { standardIo } = context;
+    const [journalKey] = options.commandArgs;
+    if (options.commandArgs.length !== 1 || journalKey === undefined || journalKey.length === 0) {
+        writeStderrLine(standardIo, `[Error] journal-skip requires one argument: <journal-key>`);
+        return false;
+    }
+
+    writeStdoutLine(standardIo, `[Command] journal-skip ${journalKey}`);
+    const replicatorBeforeReplication = context.getActiveReplicator();
+    const result = await context.replicateUserInitiated({
+        trigger: "manual",
+        progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE,
+        interaction: USER_INITIATED_REPLICATION_AUTHORITY,
+    });
+    const activeReplicator = context.getActiveReplicator();
+
+    if (activeReplicator !== replicatorBeforeReplication) {
+        writeStderrLine(
+            standardIo,
+            `[Error] The active replicator changed during replication; no Journal was skipped.`
+        );
+        return false;
+    }
+    if (isReplicationCompleted(result)) {
+        writeStderrLine(standardIo, `[Error] Replication completed without a stopped Journal; no Journal was skipped.`);
+        return false;
+    }
+    if (!isLiveSyncJournalReplicator(activeReplicator)) {
+        writeStderrLine(
+            standardIo,
+            `[Error] The active replicator is not a Journal replicator; no Journal was skipped.`
+        );
+        return false;
+    }
+
+    const stoppedJournal = activeReplicator.getStoppedJournal();
+    if (!stoppedJournal) {
+        writeStderrLine(standardIo, `[Error] Replication did not stop at a Journal; no Journal was skipped.`);
+        return false;
+    }
+    if (stoppedJournal.key !== journalKey) {
+        writeStderrLine(
+            standardIo,
+            `[Error] Replication stopped at Journal '${stoppedJournal.key}' (${stoppedJournal.stage}), not '${journalKey}'; no Journal was skipped.`
+        );
+        return false;
+    }
+
+    return await activeReplicator.skipStoppedJournalAndResume(context.currentSettings(), stoppedJournal, true);
+}

src/apps/cli/commands/journalSkip.unit.spec.ts

@@ -0,0 +1,213 @@
+import { describe, expect, it, vi } from "vitest";
+import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
+import {
+    REPLICATION_COMPLETED,
+    REPLICATION_PROGRESS_PRESENTATIONS,
+    USER_INITIATED_REPLICATION_AUTHORITY,
+    replicationFailed,
+} from "@vrtmrz/livesync-commonlib/replication";
+import { runJournalSkipCommand } from "./journalSkip";
+
+const snapshot = { key: "7-docs.jsonl.gz", stage: "parse" } as const;
+const settings = {} as any;
+type ReplicationResult = Awaited<ReturnType<Parameters<typeof runJournalSkipCommand>[1]["replicateUserInitiated"]>>;
+
+function createStandardIoMock(events?: string[]) {
+    return {
+        readStdin: vi.fn(async () => ""),
+        prompt: vi.fn(async () => ""),
+        writeStdout: vi.fn((_chunk: string | Uint8Array) => events?.push("stdout")),
+        writeStderr: vi.fn((_chunk: string | Uint8Array) => events?.push("stderr")),
+    } as any;
+}
+
+function createJournalReplicatorMock(stoppedJournal: unknown, events?: string[]) {
+    const replicator = Object.create(LiveSyncJournalReplicator.prototype) as any;
+    replicator.getStoppedJournal = vi.fn(() => {
+        events?.push("getStoppedJournal");
+        return stoppedJournal;
+    });
+    replicator.skipStoppedJournalAndResume = vi.fn(async () => {
+        events?.push("skipStoppedJournalAndResume");
+        return true;
+    });
+    return replicator;
+}
+
+function createDependencies(replicator: any, result?: ReplicationResult, events?: string[]) {
+    const standardIo = createStandardIoMock(events);
+    return {
+        standardIo,
+        getActiveReplicator: vi.fn(() => {
+            events?.push("getActiveReplicator");
+            return replicator;
+        }),
+        replicateUserInitiated: vi.fn(async () => {
+            events?.push("replicateUserInitiated");
+            return result ?? replicationFailed(new Error("stopped Journal"));
+        }),
+        currentSettings: vi.fn(() => {
+            events?.push("currentSettings");
+            return settings;
+        }),
+    };
+}
+
+function makeOptions(commandArgs: string[]) {
+    return { commandArgs };
+}
+
+describe("journal-skip command handler", () => {
+    it("retries replication, then skips the exact stopped snapshot in order", async () => {
+        const events: string[] = [];
+        const replicator = createJournalReplicatorMock(snapshot, events);
+        const dependencies = createDependencies(replicator, undefined, events);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(true);
+
+        expect(dependencies.replicateUserInitiated).toHaveBeenCalledWith({
+            trigger: "manual",
+            progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE,
+            interaction: USER_INITIATED_REPLICATION_AUTHORITY,
+        });
+        expect(dependencies.getActiveReplicator).toHaveBeenCalledTimes(2);
+        expect(replicator.getStoppedJournal).toHaveBeenCalledOnce();
+        expect(replicator.skipStoppedJournalAndResume).toHaveBeenCalledWith(settings, snapshot, true);
+        expect(replicator.skipStoppedJournalAndResume.mock.calls[0][1]).toBe(snapshot);
+        expect(events).toEqual([
+            "stdout",
+            "getActiveReplicator",
+            "replicateUserInitiated",
+            "getActiveReplicator",
+            "getStoppedJournal",
+            "currentSettings",
+            "skipStoppedJournalAndResume",
+        ]);
+        expect(dependencies.standardIo.writeStdout).toHaveBeenCalledWith(`[Command] journal-skip ${snapshot.key}\n`);
+        expect(dependencies.standardIo.writeStderr).not.toHaveBeenCalled();
+    });
+
+    it.each([
+        { args: [], description: "no argument" },
+        { args: [""], description: "an empty key" },
+        { args: [snapshot.key, "extra"], description: "more than one argument" },
+    ])("rejects $description before replication", async ({ args }) => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator);
+
+        await expect(runJournalSkipCommand(makeOptions(args), dependencies)).resolves.toBe(false);
+
+        expect(dependencies.standardIo.writeStdout).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] journal-skip requires one argument: <journal-key>\n"
+        );
+        expect(dependencies.getActiveReplicator).not.toHaveBeenCalled();
+        expect(dependencies.replicateUserInitiated).not.toHaveBeenCalled();
+    });
+
+    it("refuses a completed replication before reading a stopped snapshot", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator, REPLICATION_COMPLETED);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(false);
+
+        expect(replicator.getStoppedJournal).not.toHaveBeenCalled();
+        expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] Replication completed without a stopped Journal; no Journal was skipped.\n"
+        );
+    });
+
+    it.each([
+        { status: "failed", error: new Error("stopped") },
+        { status: "cancelled" },
+        { status: "blocked", reason: "not-ready" },
+        { status: "partial", detail: "incomplete" },
+    ] satisfies ReplicationResult[])("requires a stopped snapshot after a $status outcome", async (result) => {
+        const replicator = createJournalReplicatorMock(undefined);
+        const dependencies = createDependencies(replicator, result);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(false);
+
+        expect(replicator.getStoppedJournal).toHaveBeenCalledOnce();
+        expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] Replication did not stop at a Journal; no Journal was skipped.\n"
+        );
+    });
+
+    it("refuses an active replicator which is not a Journal replicator", async () => {
+        const otherReplicator = { getStoppedJournal: vi.fn(() => snapshot) };
+        const dependencies = createDependencies(otherReplicator);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(false);
+
+        expect(otherReplicator.getStoppedJournal).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] The active replicator is not a Journal replicator; no Journal was skipped.\n"
+        );
+    });
+
+    it("refuses if the active replicator changes while replication is awaited", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const replacement = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator);
+        dependencies.getActiveReplicator.mockReturnValueOnce(replicator).mockReturnValueOnce(replacement);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(false);
+
+        expect(replicator.getStoppedJournal).not.toHaveBeenCalled();
+        expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+        expect(replacement.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] The active replicator changed during replication; no Journal was skipped.\n"
+        );
+    });
+
+    it("refuses a stopped Journal whose key does not exactly match the requested key", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator);
+
+        await expect(runJournalSkipCommand(makeOptions(["8-docs.jsonl.gz"]), dependencies)).resolves.toBe(false);
+
+        expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+        expect(dependencies.standardIo.writeStderr).toHaveBeenCalledWith(
+            "[Error] Replication stopped at Journal '7-docs.jsonl.gz' (parse), not '8-docs.jsonl.gz'; no Journal was skipped.\n"
+        );
+    });
+
+    it("returns false when resuming after a successful Journal skip returns false", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        replicator.skipStoppedJournalAndResume.mockResolvedValue(false);
+        const dependencies = createDependencies(replicator);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).resolves.toBe(false);
+
+        expect(replicator.skipStoppedJournalAndResume).toHaveBeenCalledOnce();
+        expect(dependencies.standardIo.writeStderr).not.toHaveBeenCalled();
+    });
+
+    it("propagates rejected replication", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator);
+        const error = new Error("replication rejected");
+        dependencies.replicateUserInitiated.mockRejectedValue(error);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).rejects.toBe(error);
+
+        expect(dependencies.getActiveReplicator).toHaveBeenCalledOnce();
+        expect(dependencies.standardIo.writeStderr).not.toHaveBeenCalled();
+        expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled();
+    });
+
+    it("propagates a rejected skip and resume operation", async () => {
+        const replicator = createJournalReplicatorMock(snapshot);
+        const dependencies = createDependencies(replicator);
+        const error = new Error("skip rejected");
+        replicator.skipStoppedJournalAndResume.mockRejectedValue(error);
+
+        await expect(runJournalSkipCommand(makeOptions([snapshot.key]), dependencies)).rejects.toBe(error);
+
+        expect(replicator.skipStoppedJournalAndResume).toHaveBeenCalledOnce();
+    });
+});

src/apps/cli/commands/runCommand.ts

@@ -28,7 +28,7 @@ import {
     USER_INITIATED_REPLICATION_AUTHORITY,
 } from "@vrtmrz/livesync-commonlib/replication";
 import { withOwnedRemoteResource } from "@/common/ownedRemoteResource";
-import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
+import { runJournalSkipCommand } from "./journalSkip";
 import {
     isCentralRemoteAdministrationCommand,
     runCentralRemoteAdministrationCommand,
@@ -201,61 +201,13 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
     }
 
     if (options.command === "journal-skip") {
-        const [journalKey] = options.commandArgs;
+        return await runJournalSkipCommand(options, {
-        if (options.commandArgs.length !== 1 || journalKey === undefined || journalKey.length === 0) {
+            standardIo,
-            writeStderrLine(standardIo, `[Error] journal-skip requires one argument: <journal-key>`);
+            getActiveReplicator: () => core.services.replicator.getActiveReplicator(),
-            return false;
+            replicateUserInitiated: (replicationOptions) =>
-        }
+                core.services.replication.replicateUserInitiated(replicationOptions),
-
+            currentSettings: () => core.services.setting.currentSettings(),
-        writeStdoutLine(standardIo, `[Command] journal-skip ${journalKey}`);
-        const replicatorBeforeReplication = core.services.replicator.getActiveReplicator();
-        const result = await core.services.replication.replicateUserInitiated({
-            trigger: "manual",
-            progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE,
-            interaction: USER_INITIATED_REPLICATION_AUTHORITY,
         });
-        const activeReplicator = core.services.replicator.getActiveReplicator();
-
-        if (activeReplicator !== replicatorBeforeReplication) {
-            writeStderrLine(
-                standardIo,
-                `[Error] The active replicator changed during replication; no Journal was skipped.`
-            );
-            return false;
-        }
-        if (isReplicationCompleted(result)) {
-            writeStderrLine(
-                standardIo,
-                `[Error] Replication completed without a stopped Journal; no Journal was skipped.`
-            );
-            return false;
-        }
-        if (!(activeReplicator instanceof LiveSyncJournalReplicator)) {
-            writeStderrLine(
-                standardIo,
-                `[Error] The active replicator is not a Journal replicator; no Journal was skipped.`
-            );
-            return false;
-        }
-
-        const stoppedJournal = activeReplicator.getStoppedJournal();
-        if (!stoppedJournal) {
-            writeStderrLine(standardIo, `[Error] Replication did not stop at a Journal; no Journal was skipped.`);
-            return false;
-        }
-        if (stoppedJournal.key !== journalKey) {
-            writeStderrLine(
-                standardIo,
-                `[Error] Replication stopped at Journal '${stoppedJournal.key}' (${stoppedJournal.stage}), not '${journalKey}'; no Journal was skipped.`
-            );
-            return false;
-        }
-
-        return await activeReplicator.skipStoppedJournalAndResume(
-            core.services.setting.currentSettings(),
-            stoppedJournal,
-            true
-        );
     }
 
     if (options.command === "p2p-peers") {

src/apps/cli/package.json

@@ -12,7 +12,7 @@
         "buildRun": "npm run build && npm run cli --",
         "build:docker": "docker build -f Dockerfile -t livesync-cli ../../..",
         "check": "tsc -p tsconfig.json",
-        "test:unit": "cd ../../.. && npx vitest run --config vitest.config.unit.ts src/apps/cli/main.unit.spec.ts src/apps/cli/main.bootstrap.unit.spec.ts src/apps/cli/settingsPersistence.unit.spec.ts src/apps/cli/runtimeState.unit.spec.ts src/apps/cli/services/NodeKeyValueDBService.unit.spec.ts src/apps/cli/services/NodeLocalStorage.unit.spec.ts src/apps/cli/managers/CLIStorageEventManagerAdapter.unit.spec.ts src/apps/cli/commands/utils.unit.spec.ts src/apps/cli/commands/runCommand.unit.spec.ts src/apps/cli/commands/daemonCommand.unit.spec.ts src/apps/cli/commands/p2p.unit.spec.ts src/apps/cli/deploy/install.unit.spec.ts src/apps/cli/adapters/NodeFileSystemAdapter.unit.spec.ts",
+        "test:unit": "cd ../../.. && npx vitest run --config vitest.config.unit.ts src/apps/cli/main.unit.spec.ts src/apps/cli/main.bootstrap.unit.spec.ts src/apps/cli/settingsPersistence.unit.spec.ts src/apps/cli/runtimeState.unit.spec.ts src/apps/cli/services/NodeKeyValueDBService.unit.spec.ts src/apps/cli/services/NodeLocalStorage.unit.spec.ts src/apps/cli/managers/CLIStorageEventManagerAdapter.unit.spec.ts src/apps/cli/commands/utils.unit.spec.ts src/apps/cli/commands/runCommand.unit.spec.ts src/apps/cli/commands/journalSkip.unit.spec.ts src/apps/cli/commands/daemonCommand.unit.spec.ts src/apps/cli/commands/p2p.unit.spec.ts src/apps/cli/deploy/install.unit.spec.ts src/apps/cli/adapters/NodeFileSystemAdapter.unit.spec.ts",
         "test:e2e:two-vaults": "bash test/test-e2e-two-vaults-with-docker-linux.sh",
         "test:e2e:two-vaults:common": "bash test/test-e2e-two-vaults-common.sh",
         "test:e2e:two-vaults:matrix": "bash test/test-e2e-two-vaults-matrix.sh",

src/modules/features/SettingDialogue/PaneMaintenance.ts

@@ -9,7 +9,7 @@ import { LOG_LEVEL_NOTICE, Logger } from "@vrtmrz/livesync-commonlib/compat/comm
 import { FlagFilesHumanReadable, FlagFilesOriginal } from "@vrtmrz/livesync-commonlib/compat/common/types";
 import { fireAndForget } from "@vrtmrz/livesync-commonlib/compat/common/utils";
 import { LiveSyncCouchDBReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator";
-import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
+import { isLiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator";
 import type { StoppedJournal } from "@vrtmrz/livesync-commonlib/compat/replication/journal/JournalSyncTypes";
 import { LiveSyncSetting as Setting } from "./LiveSyncSetting.ts";
 import type { ObsidianLiveSyncSettingTab } from "./ObsidianLiveSyncSettingTab";
@@ -154,6 +154,9 @@ export function paneMaintenance(
             )
             .addOnUpdate(this.onlyOnCouchDB);
 
+        // Offer manual recovery after Journal receipt stops on a failure. Normal retries
+        // revisit the same Journal; this action confirms and skips only that Journal,
+        // resumes receipt, and still stops at the next Journal failure.
         new Setting(paneEl)
             .setName($msg("Skip stopped Journal and resume"))
             .setDesc(
@@ -167,7 +170,7 @@ export function paneMaintenance(
                     .setDisabled(false)
                     .onClick(async () => {
                         const replicator = this.core.replicator;
-                        if (!(replicator instanceof LiveSyncJournalReplicator)) {
+                        if (!isLiveSyncJournalReplicator(replicator)) {
                             new Notice(
                                 $msg(
                                     "No stopped Journal is available. Retry replication to identify the stopped Journal."

src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts

@@ -51,9 +51,13 @@ vi.mock("@vrtmrz/livesync-commonlib/compat/common/utils", () => ({
 vi.mock("@vrtmrz/livesync-commonlib/compat/replication/couchdb/LiveSyncReplicator", () => ({
     LiveSyncCouchDBReplicator: class {},
 }));
-vi.mock("@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator", () => ({
+vi.mock("@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator", () => {
-    LiveSyncJournalReplicator: class {},
+    class LiveSyncJournalReplicator {}
-}));
+    return {
+        LiveSyncJournalReplicator,
+        isLiveSyncJournalReplicator: (value: unknown) => value instanceof LiveSyncJournalReplicator,
+    };
+});
 vi.mock("./LiveSyncSetting.ts", () => ({
     LiveSyncSetting: class {
         name = "";