diff --git a/src/apps/cli/commands/journalSkip.ts b/src/apps/cli/commands/journalSkip.ts new file mode 100644 index 00000000..2bbe2c7f --- /dev/null +++ b/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; + +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 { + 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: `); + 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); +} diff --git a/src/apps/cli/commands/journalSkip.unit.spec.ts b/src/apps/cli/commands/journalSkip.unit.spec.ts new file mode 100644 index 00000000..00a11475 --- /dev/null +++ b/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[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: \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(); + }); +}); diff --git a/src/apps/cli/commands/runCommand.ts b/src/apps/cli/commands/runCommand.ts index b932d4f7..e5fd166f 100644 --- a/src/apps/cli/commands/runCommand.ts +++ b/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; - if (options.commandArgs.length !== 1 || journalKey === undefined || journalKey.length === 0) { - writeStderrLine(standardIo, `[Error] journal-skip requires one argument: `); - return false; - } - - 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, + return await runJournalSkipCommand(options, { + standardIo, + getActiveReplicator: () => core.services.replicator.getActiveReplicator(), + replicateUserInitiated: (replicationOptions) => + core.services.replication.replicateUserInitiated(replicationOptions), + currentSettings: () => core.services.setting.currentSettings(), }); - 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") { diff --git a/src/apps/cli/package.json b/src/apps/cli/package.json index 4e2ebe8e..eedb6d37 100644 --- a/src/apps/cli/package.json +++ b/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", diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.ts b/src/modules/features/SettingDialogue/PaneMaintenance.ts index da962321..b2384dbf 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.ts +++ b/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." diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts b/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts index 53d9767f..f96bbb62 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts +++ b/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", () => ({ - LiveSyncJournalReplicator: class {}, -})); +vi.mock("@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator", () => { + class LiveSyncJournalReplicator {} + return { + LiveSyncJournalReplicator, + isLiveSyncJournalReplicator: (value: unknown) => value instanceof LiveSyncJournalReplicator, + }; +}); vi.mock("./LiveSyncSetting.ts", () => ({ LiveSyncSetting: class { name = "";