From a94c3fef90acd1a559c9cf760b5141ed69cd7229 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Wed, 4 Nov 2026 15:25:30 +0000 Subject: [PATCH 1/5] Add explicit stopped-Journal recovery to Obsidian and CLI Offer a confirmation in Maintenance and the journal-skip command to skip only the current stopped Journal and resume receipt. Keep Commonlib notices visible on stderr during ordinary CLI use. Document the missing-content risk and replay procedure, and cover cancellation, stale failures, subsequent failures, and receipt with unit and plain and encrypted CLI and Obsidian end-to-end tests. Requires the Commonlib stopped-Journal recovery API. Adopt its exact published package version before integrating this consumer change. --- docs/design_docs_of_journalsync_2nd.md | 10 + docs/settings.md | 20 +- docs/troubleshooting.md | 23 ++ src/apps/cli/README.md | 25 ++ src/apps/cli/commands/runCommand.ts | 64 ++- src/apps/cli/commands/runCommand.unit.spec.ts | 129 +++++- src/apps/cli/commands/types.ts | 2 + src/apps/cli/main.log-handler.unit.spec.ts | 69 ++++ src/apps/cli/main.ts | 15 +- src/apps/cli/main.unit.spec.ts | 10 + .../messages/LiveSyncProvisionalMessages.ts | 12 + .../SettingDialogue/PaneMaintenance.ts | 70 ++++ .../PaneMaintenance.unit.spec.ts | 218 ++++++++++- test/e2e-cli/ci-tasks.json | 3 +- test/e2e-cli/tasks.json | 350 +++++++---------- test/e2e-cli/test-journal-receive-failure.ts | 369 ++++++++++++++++++ .../scripts/journal-nfc-resend.ts | 158 +++++++- updates.md | 8 + 18 files changed, 1321 insertions(+), 234 deletions(-) create mode 100644 test/e2e-cli/test-journal-receive-failure.ts diff --git a/docs/design_docs_of_journalsync_2nd.md b/docs/design_docs_of_journalsync_2nd.md index 8150da9a..c285520a 100644 --- a/docs/design_docs_of_journalsync_2nd.md +++ b/docs/design_docs_of_journalsync_2nd.md @@ -1,5 +1,15 @@ ## The design document of the Journal Replicator 2nd Edition +Implementation note: the receive-failure correction and single-Journal skip +action are unreleased. Their owning contract is Commonlib's +[Journal receive failures and recovery](https://github.com/vrtmrz/livesync-commonlib/blob/main/docs/journal-receive-recovery.md). +The receiver acknowledges successful processing, stops on failure by default, +and retries from that Journal with the same database. Explicit skipping does +not establish Vault reflection success. The host action names one stopped +Journal and its failure stage, requires confirmation, and resumes receipt after +skipping that Journal; the next failure still stops. The [settings reference](settings.md#skip-stopped-journal-and-resume) +describes the host control and replay operation. + ### Goal - Build a robust and memory-efficient replication foundation that decouples the physical storage layer by leveraging the Web Streams API. - Maintain strict compliance with the data consistency and replication protocols of CouchDB/PouchDB. diff --git a/docs/settings.md b/docs/settings.md index 46f76edd..a915638f 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -1273,9 +1273,27 @@ Disables all synchronisation and restart. Explicitly resend all locally available chunks to the remote. This is a recovery and maintenance operation; ordinary replication does not pre-send every chunk. +#### Skip stopped Journal and resume + +Object Storage only. If a stopped Journal is not available, restart replication +normally with the same local database so the receive failure identifies its key +and stage. Confirming this action skips only that Journal and resumes receipt; +it does not skip the rest of the remote history. The next Journal failure still +stops replication. + +Skipping may leave content or history missing. Normal replication does not +retry a Journal which has been explicitly skipped. After correcting the cause, +use **Reset journal received history** to replay retained remote Journals, +except those sent by this device. It cannot restore Journals which are no +longer available remotely. This action does not add a persistent ignore setting. + #### Reset journal received history -Initialise journal received history. On the next sync, every item except this device sent will be downloaded again. +Clear this device's received Journal history and known-document transfer +records. The next sync downloads the retained remote Journals again, except +those sent by this device. The local database, Vault files, and sent history +are retained. Recovery requires intact remote data and a working local +database; resetting history does not recreate missing remote content. #### Reset journal sent history diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index c68fe8fc..8da3a3d5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -47,6 +47,29 @@ Do not switch to P2P or reset the database as the first response. Check: If the remote is healthy but one device's local database is not, use [Reset Synchronisation on This Device](recovery.md#reset-synchronisation-on-this-device) only after backing up unsynchronised local files. +## Object Storage stops at a Journal receive failure + +The receiver stops when a Journal cannot be processed or its records cannot be +saved. Correct the reported cause, then retry replication with the same local +database; the failed Journal is retried. If the stopped Journal is no longer +available to the active replicator, run normal replication again to identify +the current failure. + +If the Journal cannot be recovered and missing content or history is +acceptable, open **Maintenance → Syncing → Skip stopped Journal and resume**. +The confirmation names the exact Journal key and failure stage. Confirming +skips only that Journal and resumes receipt; the next Journal failure stops +replication again. Use this only when the missing content or history is +acceptable. See the [setting reference](settings.md#skip-stopped-journal-and-resume). + +**Reset journal received history** replays remote Journals that are still +available, except those sent by this device. It retains local data and this +device's sent history, and cannot restore remote Journals which are missing. +Check the resulting file contents and retained history; a completed receive +after skipping does not prove that every Vault file is readable. CLI Fetch +replaces its local database and has separate [interrupted reconstruction +rules](../src/apps/cli/README.md#interrupted-reconstruction-and-older-runtime-state). + ## An unchanged file appears as a conflict after restart If the log says `Preserved unsynchronised local changes as a conflict`, keep both versions available until you have checked their contents. The message means that LiveSync could not prove that the Vault file was unchanged; it does not establish that you edited the file. Use **Inspect conflicts and file/database differences** in Hatch to review the current branches before selecting a version. diff --git a/src/apps/cli/README.md b/src/apps/cli/README.md index 3b176a5e..99614627 100644 --- a/src/apps/cli/README.md +++ b/src/apps/cli/README.md @@ -80,6 +80,7 @@ livesync-cli [database-path] [command] [args...] ### Commands - `sync`: Run one replication cycle with the remote CouchDB. +- `journal-skip `: Retry replication, skip only the exact currently stopped Object Storage Journal, and resume. - `mirror [vault-path]`: Bidirectional sync between the local database and a local directory (**the actual vault**). - If `vault-path` is provided, the CLI will synchronise the database with files in the vault directory. - If `vault-path` is omitted, it defaults to `database-path` (compatibility mode). @@ -108,6 +109,27 @@ livesync-cli [database-path] [command] [args...] Remote-administration commands verify the resulting milestone state through the selected provider. The existing `[Verification]` lines remain suitable for scripts which inspect command output, while the default exit code now reflects whether that verification succeeded. +### Object Storage receive failures + +`sync` stops at a Journal which cannot be processed or whose records cannot be +saved. After correcting the cause, run `sync` again with the same database and +settings to retry it. Existing partial writes are retained. `fetch --force` +replaces the local database instead and follows the reconstruction rules below. + +If the Journal cannot be recovered and missing content or history is +acceptable, run `journal-skip `. The CLI first performs normal +replication with the retained local database. It skips only when that attempt +stops at the exact supplied key; if replication succeeds or stops at another +Journal, the command refuses the explicit skip. A successful normal attempt +can still acknowledge Journals which it processed successfully. +After the matching Journal is skipped, replication resumes. Any later failed +Journal still stops, so each additional skip requires a separate command with +that Journal's key. Normal identity, connectivity, authentication, and +readiness checks still apply. This action does not persist a setting for future +replication. Review the resulting file contents and retained history; a +completed receive does not guarantee that later file reflection or reads will +succeed. + ### Interrupted reconstruction and older runtime state An interrupted `fetch` or `rebuild` leaves a record in @@ -146,6 +168,7 @@ scopes do not make concurrent writers to shared device settings supported. ```bash # Basic sync with remote livesync-cli ./my-db sync +livesync-cli ./my-db journal-skip 7-docs.jsonl.gz # Mirroring to your actual Obsidian vault livesync-cli ./my-db mirror /path/to/obsidian-vault @@ -475,6 +498,8 @@ Commands: daemon (default) Run mirror scan then continuously sync CouchDB <-> local filesystem init-settings [path] Create unconfigured settings JSON with the new-Vault recommendations sync Run one replication cycle and exit + journal-skip + Retry replication, skip only the matching stopped Journal, and resume p2p-peers Show discovered peers as [peer] p2p-sync Synchronise with specified peer-id or peer-name p2p-host Start P2P host mode and wait until interrupted (Ctrl+C) diff --git a/src/apps/cli/commands/runCommand.ts b/src/apps/cli/commands/runCommand.ts index 2c6d1174..b932d4f7 100644 --- a/src/apps/cli/commands/runCommand.ts +++ b/src/apps/cli/commands/runCommand.ts @@ -28,6 +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 { isCentralRemoteAdministrationCommand, runCentralRemoteAdministrationCommand, @@ -45,10 +46,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext writeStderrLine(standardIo, `[Error] ${forceRequirementError}`); return false; } - if ( - options.command === "fetch" && - isRemediationModeActive(core.services.setting.currentSettings()) - ) { + if (options.command === "fetch" && isRemediationModeActive(core.services.setting.currentSettings())) { writeStderrLine( standardIo, `[Error] The fetch command cannot run while remediation mode is active. Clear the modification-time limit before fetching.` @@ -202,6 +200,64 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext return isReplicationCompleted(result); } + 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, + }); + 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") { if (options.commandArgs.length < 1) { throw new Error("p2p-peers requires one argument: "); diff --git a/src/apps/cli/commands/runCommand.unit.spec.ts b/src/apps/cli/commands/runCommand.unit.spec.ts index 34de0716..ac987478 100644 --- a/src/apps/cli/commands/runCommand.unit.spec.ts +++ b/src/apps/cli/commands/runCommand.unit.spec.ts @@ -11,6 +11,7 @@ import { import { describe, expect, it, vi, beforeEach, afterEach } from "vitest"; import { runCommand } from "./runCommand"; import type { CLIOptions } from "./types"; +import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator"; import { CENTRAL_REMOTE_ADMINISTRATION_ACTIONS, CENTRAL_REMOTE_ADMINISTRATION_FAILURE_REASONS, @@ -20,6 +21,7 @@ import { CENTRAL_COMPATIBILITY_REJECTION_REASONS, REPLICATION_COMPLETED, REPLICATION_PROGRESS_PRESENTATIONS, + USER_INITIATED_REPLICATION_AUTHORITY, replicationFailed, } from "@vrtmrz/livesync-commonlib/replication"; @@ -137,6 +139,17 @@ function makeOptions(command: CLIOptions["command"], commandArgs: string[]): CLI }; } +function makeJournalSkipOptions(journalKey: string): CLIOptions { + return makeOptions("journal-skip", [journalKey]); +} + +function createJournalReplicatorMock(stoppedJournal: unknown) { + const replicator = Object.create(LiveSyncJournalReplicator.prototype) as any; + replicator.getStoppedJournal = vi.fn(() => stoppedJournal); + replicator.skipStoppedJournalAndResume = vi.fn(async () => true); + return replicator; +} + async function createSetupURI(passphrase: string): Promise { const settings = { ...DEFAULT_SETTINGS, @@ -354,6 +367,108 @@ describe("runCommand abnormal cases", () => { ); }); + describe("journal-skip command", () => { + const snapshot = { key: "7-docs.jsonl.gz", stage: "parse" } as const; + + it("skips only the exact stopped Journal after a normal user-initiated replication", async () => { + const core = createCoreMock(); + const replicator = createJournalReplicatorMock(snapshot); + core.services.replicator.getActiveReplicator.mockReturnValue(replicator); + core.services.replication.replicateUserInitiated.mockResolvedValue( + replicationFailed(new Error("stopped Journal")) + ); + + await expect(runCommand(makeJournalSkipOptions(snapshot.key), { ...context, core })).resolves.toBe(true); + + expect(core.services.replication.replicateUserInitiated).toHaveBeenCalledWith( + expect.objectContaining({ + trigger: "manual", + progressPresentation: REPLICATION_PROGRESS_PRESENTATIONS.NOTICE, + interaction: USER_INITIATED_REPLICATION_AUTHORITY, + }) + ); + expect(replicator.getStoppedJournal).toHaveBeenCalledOnce(); + expect(replicator.skipStoppedJournalAndResume).toHaveBeenCalledWith( + core.services.setting.currentSettings(), + snapshot, + true + ); + }); + + it("refuses a key which does not match the stopped Journal", async () => { + const core = createCoreMock(); + const replicator = createJournalReplicatorMock(snapshot); + core.services.replicator.getActiveReplicator.mockReturnValue(replicator); + core.services.replication.replicateUserInitiated.mockResolvedValue( + replicationFailed(new Error("stopped Journal")) + ); + + await expect(runCommand(makeJournalSkipOptions("8-docs.jsonl.gz"), { ...context, core })).resolves.toBe( + false + ); + + expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + expect(core.services.context.standardIo.writeStderr).toHaveBeenCalledWith( + expect.stringContaining("7-docs.jsonl.gz") + ); + }); + + it("refuses when normal replication does not establish a stopped Journal", async () => { + const core = createCoreMock(); + const replicator = createJournalReplicatorMock(undefined); + core.services.replicator.getActiveReplicator.mockReturnValue(replicator); + + await expect(runCommand(makeJournalSkipOptions(snapshot.key), { ...context, core })).resolves.toBe(false); + + expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + }); + + it("refuses when the active replicator is not a Journal replicator", async () => { + const core = createCoreMock(); + core.services.replicator.getActiveReplicator.mockReturnValue({ getStoppedJournal: vi.fn(() => snapshot) }); + core.services.replication.replicateUserInitiated.mockResolvedValue( + replicationFailed(new Error("stopped Journal")) + ); + + await expect(runCommand(makeJournalSkipOptions(snapshot.key), { ...context, core })).resolves.toBe(false); + + expect(core.services.context.standardIo.writeStderr).toHaveBeenCalledWith( + expect.stringContaining("Journal replicator") + ); + }); + + it("propagates a failure while resuming after the requested Journal was skipped", async () => { + const core = createCoreMock(); + const replicator = createJournalReplicatorMock(snapshot); + replicator.skipStoppedJournalAndResume.mockResolvedValue(false); + core.services.replicator.getActiveReplicator.mockReturnValue(replicator); + core.services.replication.replicateUserInitiated.mockResolvedValue( + replicationFailed(new Error("stopped Journal")) + ); + + await expect(runCommand(makeJournalSkipOptions(snapshot.key), { ...context, core })).resolves.toBe(false); + + expect(replicator.skipStoppedJournalAndResume).toHaveBeenCalledOnce(); + }); + + it("refuses if the active replicator changes during the awaited replication", async () => { + const core = createCoreMock(); + const replicator = createJournalReplicatorMock(snapshot); + const replacement = createJournalReplicatorMock(snapshot); + core.services.replicator.getActiveReplicator + .mockReturnValueOnce(replicator) + .mockReturnValueOnce(replacement); + core.services.replication.replicateUserInitiated.mockResolvedValue( + replicationFailed(new Error("stopped Journal")) + ); + + await expect(runCommand(makeJournalSkipOptions(snapshot.key), { ...context, core })).resolves.toBe(false); + + expect(replicator.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + expect(replacement.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + }); + }); + it("reports a lock from the exact sync outcome without inspecting a replacement Replicator", async () => { const core = createCoreMock(); core.services.replication.replicateUserInitiated.mockResolvedValue( @@ -546,9 +661,9 @@ describe("runCommand abnormal cases", () => { clock.mockReturnValue(end); const outOfWindow = createCoreMock(); outOfWindow.services.context.standardIo.prompt.mockResolvedValue(passphrase); - await expect(runCommand(makeOptions("setup", [uri.trim()]), { ...context, core: outOfWindow })).rejects.toThrow( - "Cannot open Setup URI" - ); + await expect( + runCommand(makeOptions("setup", [uri.trim()]), { ...context, core: outOfWindow }) + ).rejects.toThrow("Cannot open Setup URI"); expect(outOfWindow.services.setting.applyExternalSettings).not.toHaveBeenCalled(); expect(outOfWindow.services.control.applySettings).not.toHaveBeenCalled(); } finally { @@ -559,8 +674,12 @@ describe("runCommand abnormal cases", () => { it("setup imports managed TURN through the existing encrypted URI", async () => { const core = createCoreMock(); const profiles = { - turn: { id: "turn", name: "TURN", isEncrypted: false, - uri: "sls+p2p://room?managedType=CF&managedId=turn-key&token=private-token" }, + turn: { + id: "turn", + name: "TURN", + isEncrypted: false, + uri: "sls+p2p://room?managedType=CF&managedId=turn-key&token=private-token", + }, }; const passphrase = "setup-passphrase"; const setupURI = await processSetting.encodeSettingsToSetupURI( diff --git a/src/apps/cli/commands/types.ts b/src/apps/cli/commands/types.ts index c5348a44..bd04d4bb 100644 --- a/src/apps/cli/commands/types.ts +++ b/src/apps/cli/commands/types.ts @@ -7,6 +7,7 @@ import type { ReplicationSchedulingControl } from "@/serviceFeatures/replication export type CLICommand = | "daemon" | "sync" + | "journal-skip" | "p2p-peers" | "p2p-sync" | "p2p-host" @@ -87,6 +88,7 @@ export interface CLICommandContext { export const VALID_COMMANDS = new Set([ "daemon", "sync", + "journal-skip", "p2p-peers", "p2p-sync", "p2p-host", diff --git a/src/apps/cli/main.log-handler.unit.spec.ts b/src/apps/cli/main.log-handler.unit.spec.ts index 1e7adca2..5c344a10 100644 --- a/src/apps/cli/main.log-handler.unit.spec.ts +++ b/src/apps/cli/main.log-handler.unit.spec.ts @@ -2,6 +2,16 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import type { LOG_LEVEL } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { + Logger, + LOG_LEVEL_DEBUG, + LOG_LEVEL_INFO, + LOG_LEVEL_NOTICE, + LOG_LEVEL_URGENT, + LOG_LEVEL_VERBOSE, + setGlobalLogFunction, +} from "octagonal-wheels/common/logger"; const mocks = vi.hoisted(() => ({ setLogHandler: vi.fn(), @@ -29,6 +39,18 @@ function createStandardIoMock() { }; } +async function invokeGlobalLogger(databasePath: string, message: string, level: LOG_LEVEL, debug = false) { + process.argv = ["node", "livesync-cli", databasePath, ...(debug ? ["--debug"] : []), "remote-ls"]; + const standardIo = createStandardIoMock(); + mocks.setLogHandler.mockImplementation(() => { + Logger(message, level); + throw new Error("__LOG_HANDLER_CONFIGURED__"); + }); + + await expect(main(standardIo)).rejects.toThrow("__LOG_HANDLER_CONFIGURED__"); + return standardIo; +} + describe("CLI log handler", () => { const originalArgv = process.argv.slice(); let databasePath: string; @@ -43,6 +65,7 @@ describe("CLI log handler", () => { afterEach(async () => { process.argv = originalArgv.slice(); + setGlobalLogFunction(() => {}); await rm(databasePath, { recursive: true, force: true }); }); @@ -52,4 +75,50 @@ describe("CLI log handler", () => { await expect(main(createStandardIoMock())).rejects.toThrow("__LOG_HANDLER_CONFIGURED__"); expect(mocks.setLogHandler).toHaveBeenCalledWith(expect.any(Function), true); }); + + it.each([ + ["NOTICE", LOG_LEVEL_NOTICE], + ["URGENT", LOG_LEVEL_URGENT], + ] as const)("writes Commonlib %s diagnostics to stderr by default", async (name, level) => { + const message = `Commonlib ${name} diagnostic`; + const standardIo = await invokeGlobalLogger(databasePath, message, level); + + expect(standardIo.writeStderr).toHaveBeenCalledWith(expect.stringContaining(`[${level}] ${message}\n`)); + expect(standardIo.writeStdout).not.toHaveBeenCalledWith(expect.stringContaining(message)); + }); + + it.each([ + ["DEBUG", LOG_LEVEL_DEBUG], + ["INFO", LOG_LEVEL_INFO], + ["VERBOSE", LOG_LEVEL_VERBOSE], + ] as const)("keeps Commonlib %s diagnostics hidden by default", async (name, level) => { + const message = `Commonlib ${name} diagnostic`; + const standardIo = await invokeGlobalLogger(databasePath, message, level); + + expect(standardIo.writeStderr).not.toHaveBeenCalledWith(expect.stringContaining(message)); + expect(standardIo.writeStdout).not.toHaveBeenCalledWith(expect.stringContaining(message)); + }); + + it("writes detailed Commonlib diagnostics to stderr in debug mode", async () => { + const messages = [ + ["DEBUG", LOG_LEVEL_DEBUG], + ["INFO", LOG_LEVEL_INFO], + ["VERBOSE", LOG_LEVEL_VERBOSE], + ] as const; + process.argv = ["node", "livesync-cli", databasePath, "--debug", "remote-ls"]; + const standardIo = createStandardIoMock(); + mocks.setLogHandler.mockImplementation(() => { + for (const [name, level] of messages) Logger(`Commonlib ${name} diagnostic`, level); + throw new Error("__LOG_HANDLER_CONFIGURED__"); + }); + + await expect(main(standardIo)).rejects.toThrow("__LOG_HANDLER_CONFIGURED__"); + + for (const [name, level] of messages) { + expect(standardIo.writeStderr).toHaveBeenCalledWith( + expect.stringContaining(`[${level}] Commonlib ${name} diagnostic\n`) + ); + } + expect(standardIo.writeStdout).not.toHaveBeenCalledWith(expect.stringContaining("Commonlib")); + }); }); diff --git a/src/apps/cli/main.ts b/src/apps/cli/main.ts index b41ac2d3..511e015d 100644 --- a/src/apps/cli/main.ts +++ b/src/apps/cli/main.ts @@ -108,6 +108,8 @@ Arguments: Commands: daemon (default) Run mirror scan then continuously sync CouchDB <-> local filesystem sync Run one replication cycle and exit + journal-skip + Retry replication, skip only the matching stopped Journal, and resume p2p-peers Show discovered peers as [peer]\t\t p2p-sync Sync with the specified peer-id or peer-name @@ -376,15 +378,10 @@ export async function main( const writeLine = avoidStdoutNoise ? writeStderrLine : writeStdoutLine; writeLine(standardIo, ...values); }; - if (options.debug) { - setGlobalLogFunction((msg, level) => { - writeStderrLine(standardIo, `[${level}]`, msg); - }); - } else { - setGlobalLogFunction((msg, level) => { - // NO OP, leave it to logFunction - }); - } + setGlobalLogFunction((msg, level) => { + if (!options.debug && level !== LOG_LEVEL_NOTICE && level !== LOG_LEVEL_URGENT) return; + writeStderrLine(standardIo, `[${level}]`, msg); + }); if (options.command === "init-settings") { await createDefaultSettingsFile(options, standardIo); return; diff --git a/src/apps/cli/main.unit.spec.ts b/src/apps/cli/main.unit.spec.ts index 7d730f5d..29d6fc78 100644 --- a/src/apps/cli/main.unit.spec.ts +++ b/src/apps/cli/main.unit.spec.ts @@ -72,6 +72,7 @@ describe("CLI parseArgs", () => { expect(combined).toContain("Usage:"); expect(combined).toContain("livesync-cli [options] [command-args]"); expect(combined).toContain("--compat-remote-admin-exit-zero"); + expect(combined).toContain("journal-skip "); }); it("parses p2p-peers command and timeout", () => { @@ -101,6 +102,15 @@ describe("CLI parseArgs", () => { expect(parsed.commandArgs).toEqual([]); }); + it("parses journal-skip and its exact Journal key", () => { + process.argv = ["node", "livesync-cli", "./databasePath", "journal-skip", "7-docs.jsonl.gz"]; + const parsed = parseArgs(); + + expect(parsed.databasePath).toBe("./databasePath"); + expect(parsed.command).toBe("journal-skip"); + expect(parsed.commandArgs).toEqual(["7-docs.jsonl.gz"]); + }); + it.each(["rebuild", "fetch"] as const)("parses %s and its force flag", (command) => { process.argv = ["node", "livesync-cli", "./databasePath", "--force", command]; const parsed = parseArgs(); diff --git a/src/common/messages/LiveSyncProvisionalMessages.ts b/src/common/messages/LiveSyncProvisionalMessages.ts index 4733ab44..2b14f722 100644 --- a/src/common/messages/LiveSyncProvisionalMessages.ts +++ b/src/common/messages/LiveSyncProvisionalMessages.ts @@ -7,6 +7,18 @@ * remove it from this map in the same change. */ export const liveSyncProvisionalEnglishMessages = { + "Skip stopped Journal and resume": "Skip stopped Journal and resume", + "Object Storage only. Confirm to skip the currently stopped Journal and resume replication. The confirmation names the exact Journal key and failure stage. Skipping may leave content or history missing; the next Journal failure still stops replication.": + "Object Storage only. Confirm to skip the currently stopped Journal and resume replication. The confirmation names the exact Journal key and failure stage. Skipping may leave content or history missing; the next Journal failure still stops replication.", + "Skip and resume": "Skip and resume", + "No stopped Journal is available. Retry replication to identify the stopped Journal.": + "No stopped Journal is available. Retry replication to identify the stopped Journal.", + "Skip stopped Journal '${KEY}' at failure stage '${STAGE}' and resume replication? This may leave content or history missing.": + "Skip stopped Journal '${KEY}' at failure stage '${STAGE}' and resume replication? This may leave content or history missing.", + "The active replicator changed. Retry replication to identify the stopped Journal.": + "The active replicator changed. Retry replication to identify the stopped Journal.", + "Journal receipt did not complete after this action. Check the current failure before retrying.": + "Journal receipt did not complete after this action. Check the current failure before retrying.", "File identity": "File identity", "Version 0 preserves legacy path identity; version 1 uses Unicode NFC. Physical file names stay unchanged. Update every device and stop synchronisation, then Rebuild from a complete authoritative Vault and Fetch on the other devices. With P2P, Rebuild prepares only the source device's local database. Resolve colliding file names first.": "Version 0 preserves legacy path identity; version 1 uses Unicode NFC. Physical file names stay unchanged. Update every device and stop synchronisation, then Rebuild from a complete authoritative Vault and Fetch on the other devices. With P2P, Rebuild prepares only the source device's local database. Resolve colliding file names first.", diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.ts b/src/modules/features/SettingDialogue/PaneMaintenance.ts index 096baf3c..da962321 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.ts +++ b/src/modules/features/SettingDialogue/PaneMaintenance.ts @@ -1,5 +1,6 @@ import { EVENT_REQUEST_PERFORM_GC_V3, eventHub } from "@/common/events.ts"; import { $msg } from "@/common/translation"; +import { Notice } from "@/deps.ts"; import { createCoreSettingsAfterFullReset, createEditingSettingsAfterFullReset, @@ -8,6 +9,8 @@ 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 type { StoppedJournal } from "@vrtmrz/livesync-commonlib/compat/replication/journal/JournalSyncTypes"; import { LiveSyncSetting as Setting } from "./LiveSyncSetting.ts"; import type { ObsidianLiveSyncSettingTab } from "./ObsidianLiveSyncSettingTab"; import { visibleOnly, type PageFunctions } from "./SettingPane"; @@ -151,6 +154,73 @@ export function paneMaintenance( ) .addOnUpdate(this.onlyOnCouchDB); + new Setting(paneEl) + .setName($msg("Skip stopped Journal and resume")) + .setDesc( + $msg( + "Object Storage only. Confirm to skip the currently stopped Journal and resume replication. The confirmation names the exact Journal key and failure stage. Skipping may leave content or history missing; the next Journal failure still stops replication." + ) + ) + .addButton((button) => + setButtonDestructiveState(button) + .setButtonText($msg("Skip and resume")) + .setDisabled(false) + .onClick(async () => { + const replicator = this.core.replicator; + if (!(replicator instanceof LiveSyncJournalReplicator)) { + new Notice( + $msg( + "No stopped Journal is available. Retry replication to identify the stopped Journal." + ) + ); + return; + } + const stoppedJournal: Readonly | undefined = replicator.getStoppedJournal(); + if (!stoppedJournal) { + new Notice( + $msg( + "No stopped Journal is available. Retry replication to identify the stopped Journal." + ) + ); + return; + } + const confirmed = + (await this.core.confirm.askYesNoDialog( + $msg( + "Skip stopped Journal '${KEY}' at failure stage '${STAGE}' and resume replication? This may leave content or history missing.", + { KEY: stoppedJournal.key, STAGE: stoppedJournal.stage } + ), + { + title: $msg("Skip stopped Journal and resume"), + defaultOption: "No", + } + )) === "yes"; + if (!confirmed) { + return; + } + if (this.core.replicator !== replicator) { + new Notice( + $msg( + "The active replicator changed. Retry replication to identify the stopped Journal." + ) + ); + return; + } + const resumed = await replicator.skipStoppedJournalAndResume( + this.core.settings, + stoppedJournal + ); + if (!resumed) { + new Notice( + $msg( + "Journal receipt did not complete after this action. Check the current failure before retrying." + ) + ); + } + }) + ) + .addOnUpdate(this.onlyOnMinIO); + new Setting(paneEl) .setName($msg("Reset journal received history")) .setDesc( diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts b/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts index 7f5bf8b5..53d9767f 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts +++ b/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts @@ -1,8 +1,14 @@ import { afterEach, describe, expect, it, vi } from "vitest"; const maintenanceHarness = vi.hoisted(() => ({ - createdSettings: [] as Array<{ name: string; click?: () => Promise }>, + createdSettings: [] as Array<{ + name: string; + click?: () => Promise; + toggleKey?: string; + onUpdate?: () => { visibility?: boolean }; + }>, logger: vi.fn(), + notice: vi.fn(), })); vi.mock("@/common/events.ts", () => ({ @@ -10,7 +16,19 @@ vi.mock("@/common/events.ts", () => ({ eventHub: { emitEvent: vi.fn() }, })); vi.mock("@/common/translation", () => ({ - $msg: (message: string) => message, + $msg: (message: string, replacements?: Record) => { + for (const [key, value] of Object.entries(replacements ?? {})) { + message = message.split(String.fromCharCode(36) + "{" + key + "}").join(String(value)); + } + return message; + }, +})); +vi.mock("@/deps.ts", () => ({ + Notice: class { + constructor(message: string) { + maintenanceHarness.notice(message); + } + }, })); vi.mock("@/serviceFeatures/setupObsidian/settingsReset.ts", () => ({ createCoreSettingsAfterFullReset: vi.fn(), @@ -33,10 +51,15 @@ 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("./LiveSyncSetting.ts", () => ({ LiveSyncSetting: class { name = ""; click?: () => Promise; + toggleKey?: string; + onUpdate?: () => { visibility?: boolean }; constructor() { maintenanceHarness.createdSettings.push(this); @@ -51,6 +74,11 @@ vi.mock("./LiveSyncSetting.ts", () => ({ return this; } + autoWireToggle(key: string) { + this.toggleKey = key; + return this; + } + addButton(callback: (button: this) => void) { callback(this); return this; @@ -73,7 +101,8 @@ vi.mock("./LiveSyncSetting.ts", () => ({ return this; } - addOnUpdate() { + addOnUpdate(onUpdate: () => { visibility?: boolean }) { + this.onUpdate = onUpdate; return this; } }, @@ -85,14 +114,61 @@ vi.mock("./settingComponentStyles.ts", () => ({ setButtonDestructiveState: (button: T) => button, })); +import { LiveSyncJournalReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/journal/LiveSyncJournalReplicator"; import { paneMaintenance } from "./PaneMaintenance.ts"; afterEach(() => { maintenanceHarness.createdSettings.length = 0; maintenanceHarness.logger.mockClear(); + maintenanceHarness.notice.mockClear(); vi.clearAllMocks(); }); +type TestStoppedJournal = Readonly<{ key: string; stage: string }>; +type ConfirmationOptions = { title: string; defaultOption: "No" }; +type AskYesNoDialog = (message: string, options: ConfirmationOptions) => Promise<"yes" | "no">; + +function createJournalReplicator(stoppedJournal: TestStoppedJournal | undefined, result = true) { + const getStoppedJournal = vi.fn(() => stoppedJournal); + const skipStoppedJournalAndResume = vi.fn(async (_settings: unknown, _snapshot: unknown) => result); + const replicator = Object.assign(new LiveSyncJournalReplicator({} as never), { + getStoppedJournal, + skipStoppedJournalAndResume, + }); + return { replicator, getStoppedJournal, skipStoppedJournalAndResume }; +} + +function createSyncingPane(replicator: unknown, askYesNoDialog: AskYesNoDialog = async () => "no") { + const settings = { databaseName: "test-database" }; + const onlyOnMinIO = vi.fn(() => ({ visibility: true })); + const confirmation = vi.fn(askYesNoDialog); + const host = { + core: { replicator, settings, confirm: { askYesNoDialog: confirmation } }, + createEl: vi.fn(), + getMinioJournalSyncClient: vi.fn(), + onlyOnCouchDB: vi.fn(), + onlyOnCouchDBOrMinIO: vi.fn(), + onlyOnMinIO, + services: { replication: { markLocked: vi.fn(), markUnlocked: vi.fn() } }, + }; + const addPanel = vi.fn((_parent: HTMLElement, heading: string) => ({ + then(callback: (paneEl: HTMLElement) => void) { + if (heading === "Syncing") { + callback({} as HTMLElement); + } + return Promise.resolve(); + }, + })); + + paneMaintenance.call(host as never, {} as HTMLElement, { addPanel } as never); + + const setting = maintenanceHarness.createdSettings.find(({ name }) => name === "Skip stopped Journal and resume"); + if (!setting?.click) { + throw new Error("Skip stopped Journal action was not registered"); + } + return { setting, click: setting.click, host, settings, askYesNoDialog: confirmation, onlyOnMinIO }; +} + describe("paneMaintenance", () => { it("does not announce success when the remote wipe reports failure", async () => { const updateCheckPointInfo = vi.fn(async () => undefined); @@ -180,9 +256,139 @@ describe("paneMaintenance", () => { expect(resetDatabase).toHaveBeenCalledOnce(); expect(initialiseDatabase).toHaveBeenCalledOnce(); - expect(maintenanceHarness.logger).toHaveBeenCalledWith( - "Ui.Common.LocalDatabaseInitialisationFailed", - "notice" + expect(maintenanceHarness.logger).toHaveBeenCalledWith("Ui.Common.LocalDatabaseInitialisationFailed", "notice"); + }); + + it("removes the broken journal toggle and preserves received history reset semantics", async () => { + const checkpoint = { + receivedFiles: new Set(["received-file"]), + knownIDs: new Set(["known-id"]), + sentIDs: new Set(["sent-id"]), + sentFiles: new Set(["sent-file"]), + lastLocalSeq: 23, + }; + let updatedCheckpoint: typeof checkpoint | undefined; + const updateCheckPointInfo = vi.fn(async (update: (info: typeof checkpoint) => typeof checkpoint) => { + updatedCheckpoint = update(checkpoint); + }); + const onlyOnMinIO = vi.fn(() => ({ visibility: true })); + const addPanel = vi.fn((_parent: HTMLElement, heading: string) => ({ + then(callback: (paneEl: HTMLElement) => void) { + if (heading === "Syncing") { + callback({} as HTMLElement); + } + return Promise.resolve(); + }, + })); + const host = { + core: { replicator: {}, storageAccess: {} }, + createEl: vi.fn(), + getMinioJournalSyncClient: vi.fn(() => ({ updateCheckPointInfo })), + onlyOnCouchDB: vi.fn(), + onlyOnCouchDBOrMinIO: vi.fn(), + onlyOnMinIO, + services: { replication: { markLocked: vi.fn(), markUnlocked: vi.fn() } }, + }; + + paneMaintenance.call(host as never, {} as HTMLElement, { addPanel } as never); + + expect(maintenanceHarness.createdSettings.some(({ toggleKey }) => toggleKey === "ignoreBrokenJournals")).toBe( + false + ); + const resetReceivedHistory = maintenanceHarness.createdSettings.find( + ({ name }) => name === "Reset journal received history" + ); + if (!resetReceivedHistory?.click) { + throw new Error("Reset journal received history action was not registered"); + } + await resetReceivedHistory.click(); + + expect(updateCheckPointInfo).toHaveBeenCalledOnce(); + expect(updatedCheckpoint).toEqual({ + ...checkpoint, + receivedFiles: new Set(), + knownIDs: new Set(), + }); + expect(maintenanceHarness.logger).toHaveBeenCalledWith("Journal received history has been cleared.", "notice"); + }); + + it("asks the user to retry when there is no stopped Journal", async () => { + const { replicator, getStoppedJournal, skipStoppedJournalAndResume } = createJournalReplicator(undefined); + const rendered = createSyncingPane(replicator); + + expect(rendered.setting.onUpdate).toBe(rendered.onlyOnMinIO); + await rendered.click(); + + expect(getStoppedJournal).toHaveBeenCalledOnce(); + expect(rendered.askYesNoDialog).not.toHaveBeenCalled(); + expect(skipStoppedJournalAndResume).not.toHaveBeenCalled(); + expect(maintenanceHarness.notice).toHaveBeenCalledWith( + "No stopped Journal is available. Retry replication to identify the stopped Journal." + ); + }); + + it("does not skip the stopped Journal when confirmation is cancelled", async () => { + const stoppedJournal = { key: "journal/one", stage: "save-records" } as const; + const { replicator, skipStoppedJournalAndResume } = createJournalReplicator(stoppedJournal); + const rendered = createSyncingPane(replicator); + + await rendered.click(); + + expect(rendered.askYesNoDialog).toHaveBeenCalledOnce(); + const call = rendered.askYesNoDialog.mock.calls[0]; + if (!call) { + throw new Error("Stopped Journal confirmation was not requested"); + } + const [message, options] = call; + expect(message).toContain(stoppedJournal.key); + expect(message).toContain(stoppedJournal.stage); + expect(options).toEqual({ title: "Skip stopped Journal and resume", defaultOption: "No" }); + expect(skipStoppedJournalAndResume).not.toHaveBeenCalled(); + }); + + it("skips and resumes with the exact confirmed stopped Journal snapshot", async () => { + const stoppedJournal = { key: "journal/one", stage: "save-records" } as const; + const { replicator, skipStoppedJournalAndResume } = createJournalReplicator(stoppedJournal); + const confirm = vi.fn(async () => "yes" as const); + const rendered = createSyncingPane(replicator, confirm); + + await rendered.click(); + + expect(skipStoppedJournalAndResume).toHaveBeenCalledOnce(); + expect(skipStoppedJournalAndResume.mock.calls[0]?.[0]).toBe(rendered.settings); + expect(skipStoppedJournalAndResume.mock.calls[0]?.[1]).toBe(stoppedJournal); + }); + + it("does not skip after the active replicator changes during confirmation", async () => { + const stoppedJournal = { key: "journal/one", stage: "save-records" } as const; + const active = createJournalReplicator(stoppedJournal); + const replacement = createJournalReplicator(stoppedJournal); + const confirm = vi.fn(async () => "yes" as const); + const rendered = createSyncingPane(active.replicator, confirm); + confirm.mockImplementation(async () => { + rendered.host.core.replicator = replacement.replicator; + return "yes"; + }); + + await rendered.click(); + + expect(active.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + expect(replacement.skipStoppedJournalAndResume).not.toHaveBeenCalled(); + expect(maintenanceHarness.notice).toHaveBeenCalledWith( + "The active replicator changed. Retry replication to identify the stopped Journal." + ); + }); + + it("reports when receipt does not complete after skip and resume", async () => { + const stoppedJournal = { key: "journal/one", stage: "save-records" } as const; + const { replicator, skipStoppedJournalAndResume } = createJournalReplicator(stoppedJournal, false); + const rendered = createSyncingPane(replicator, async () => "yes"); + + await rendered.click(); + + expect(skipStoppedJournalAndResume).toHaveBeenCalledOnce(); + expect(maintenanceHarness.notice).toHaveBeenCalledWith( + "Journal receipt did not complete after this action. Check the current failure before retrying." ); }); }); diff --git a/test/e2e-cli/ci-tasks.json b/test/e2e-cli/ci-tasks.json index 4848c5ae..0931f3f3 100644 --- a/test/e2e-cli/ci-tasks.json +++ b/test/e2e-cli/ci-tasks.json @@ -16,5 +16,6 @@ "test:e2e-matrix:minio-enc0", "test:e2e-matrix:minio-enc1", "test:fetch-database-continuity", - "test:journal-chunk-recovery" + "test:journal-chunk-recovery", + "test:journal-receive-failure" ] diff --git a/test/e2e-cli/tasks.json b/test/e2e-cli/tasks.json index ee62a00e..f773f442 100644 --- a/test/e2e-cli/tasks.json +++ b/test/e2e-cli/tasks.json @@ -1,209 +1,145 @@ { - "test:helpers": { - "test": [ - "helpers/process.test.ts", - "test-ci-task-contract.ts" - ] - }, - "test": { - "tasks": [ - "test:ci" - ] - }, - "test:ci": { - "script": "run-ci-suite.ts" - }, - "test:p2p:compose": { - "script": "run-compose-p2p.ts" - }, - "test:local": { - "tasks": [ - "test:setup-put-cat", - "test:mirror", - "test:daemon" - ] - }, - "test:daemon": { - "test": [ - "test-daemon.ts" - ] - }, - "test:daemon-startup": { - "test": [ - "test-daemon-startup.ts" - ] - }, - "test:decoupled-vault": { - "test": [ - "test-decoupled-vault.ts" - ] - }, - "test:remote-commands": { - "test": [ - "test-remote-commands.ts" - ] - }, - "test:settings-writeback": { - "test": [ - "test-settings-writeback.ts" - ] - }, - "test:fetch-database-continuity": { - "test": [ - "test-fetch-database-continuity.ts" - ] - }, - "test:remote-administration-exit-codes": { - "test": [ - "test-remote-administration-exit-codes.ts" - ] - }, - "test:push-pull": { - "test": [ - "test-push-pull.ts" - ] - }, - "test:setup-put-cat": { - "test": [ - "test-setup-put-cat.ts" - ] - }, - "test:mirror": { - "test": [ - "test-mirror.ts" - ] - }, - "test:sync-two-local": { - "test": [ - "test-sync-two-local-databases.ts" - ] - }, - "test:sync-locked-remote": { - "test": [ - "test-sync-locked-remote.ts" - ] - }, - "test:p2p-host": { - "test": [ - "test-p2p-host.ts" - ] - }, - "test:p2p-peers": { - "test": [ - "test-p2p-peers-local-relay.ts" - ] - }, - "test:p2p-sync": { - "test": [ - "test-p2p-sync.ts" - ] - }, - "test:p2p-automatic-host": { - "test": [ - "test-p2p-automatic-host.ts" - ] - }, - "test:p2p-replacement": { - "test": [ - "test-p2p-replicator-replacement.ts" - ] - }, - "test:p2p-relay-disconnect": { - "test": [ - "test-p2p-relay-disconnect.ts" - ] - }, - "test:p2p:ci": { - "test": [ - "test-p2p-sync.ts", - "test-p2p-automatic-host.ts", - "test-p2p-replicator-replacement.ts", - "test-p2p-relay-disconnect.ts" - ] - }, - "test:p2p-three-nodes": { - "test": [ - "test-p2p-three-nodes-conflict.ts" - ] - }, - "test:p2p-upload-download": { - "test": [ - "test-p2p-upload-download-repro.ts" - ] - }, - "test:benchmark-contract": { - "test": [ - "test-benchmark-contract.ts" - ] - }, - "bench:p2p": { - "script": "bench-p2p.ts" - }, - "bench:couchdb": { - "script": "bench-couchdb.ts" - }, - "bench:compression": { - "script": "bench-compression.ts" - }, - "bench:cases": { - "script": "bench-network-cases.ts" - }, - "bench:latency-sweep": { - "script": "bench-latency-sweep.ts" - }, - "bench:p2p-split-node": { - "script": "bench-p2p-split-node.ts" - }, - "bench:item1": { - "command": "bash", - "args": [ - "./bench-run-item1.sh" - ] - }, - "bench:item1:full": { - "command": "bash", - "args": [ - "-lc", - "BENCH_MD_FILE_COUNT=1500 BENCH_MD_MIN_SIZE_BYTES=1024 BENCH_MD_MAX_SIZE_BYTES=20480 BENCH_BIN_FILE_COUNT=500 BENCH_BIN_SIZE_BYTES=102400 BENCH_COUCHDB_RTT_MS=50 bash ./bench-run-item1.sh" - ] - }, - "test:e2e-couchdb": { - "test": [ - "test-e2e-two-vaults-couchdb.ts" - ] - }, - "test:journal-chunk-recovery": { - "test": [ - "test-journal-chunk-recovery.ts" - ] - }, - "test:e2e-matrix": { - "test": [ - "test-e2e-two-vaults-matrix.ts" - ] - }, - "test:e2e-matrix:couchdb-enc0": { - "test": [ - "test-e2e-two-vaults-matrix.ts" - ], - "filter": "e2e matrix: COUCHDB-enc0" - }, - "test:e2e-matrix:couchdb-enc1": { - "test": [ - "test-e2e-two-vaults-matrix.ts" - ], - "filter": "e2e matrix: COUCHDB-enc1" - }, - "test:e2e-matrix:minio-enc0": { - "test": [ - "test-e2e-two-vaults-matrix.ts" - ], - "filter": "e2e matrix: MINIO-enc0" - }, - "test:e2e-matrix:minio-enc1": { - "test": [ - "test-e2e-two-vaults-matrix.ts" - ], - "filter": "e2e matrix: MINIO-enc1" - } + "test:helpers": { + "test": ["helpers/process.test.ts", "test-ci-task-contract.ts"] + }, + "test": { + "tasks": ["test:ci"] + }, + "test:ci": { + "script": "run-ci-suite.ts" + }, + "test:p2p:compose": { + "script": "run-compose-p2p.ts" + }, + "test:local": { + "tasks": ["test:setup-put-cat", "test:mirror", "test:daemon"] + }, + "test:daemon": { + "test": ["test-daemon.ts"] + }, + "test:daemon-startup": { + "test": ["test-daemon-startup.ts"] + }, + "test:decoupled-vault": { + "test": ["test-decoupled-vault.ts"] + }, + "test:remote-commands": { + "test": ["test-remote-commands.ts"] + }, + "test:settings-writeback": { + "test": ["test-settings-writeback.ts"] + }, + "test:fetch-database-continuity": { + "test": ["test-fetch-database-continuity.ts"] + }, + "test:remote-administration-exit-codes": { + "test": ["test-remote-administration-exit-codes.ts"] + }, + "test:push-pull": { + "test": ["test-push-pull.ts"] + }, + "test:setup-put-cat": { + "test": ["test-setup-put-cat.ts"] + }, + "test:mirror": { + "test": ["test-mirror.ts"] + }, + "test:sync-two-local": { + "test": ["test-sync-two-local-databases.ts"] + }, + "test:sync-locked-remote": { + "test": ["test-sync-locked-remote.ts"] + }, + "test:p2p-host": { + "test": ["test-p2p-host.ts"] + }, + "test:p2p-peers": { + "test": ["test-p2p-peers-local-relay.ts"] + }, + "test:p2p-sync": { + "test": ["test-p2p-sync.ts"] + }, + "test:p2p-automatic-host": { + "test": ["test-p2p-automatic-host.ts"] + }, + "test:p2p-replacement": { + "test": ["test-p2p-replicator-replacement.ts"] + }, + "test:p2p-relay-disconnect": { + "test": ["test-p2p-relay-disconnect.ts"] + }, + "test:p2p:ci": { + "test": [ + "test-p2p-sync.ts", + "test-p2p-automatic-host.ts", + "test-p2p-replicator-replacement.ts", + "test-p2p-relay-disconnect.ts" + ] + }, + "test:p2p-three-nodes": { + "test": ["test-p2p-three-nodes-conflict.ts"] + }, + "test:p2p-upload-download": { + "test": ["test-p2p-upload-download-repro.ts"] + }, + "test:benchmark-contract": { + "test": ["test-benchmark-contract.ts"] + }, + "bench:p2p": { + "script": "bench-p2p.ts" + }, + "bench:couchdb": { + "script": "bench-couchdb.ts" + }, + "bench:compression": { + "script": "bench-compression.ts" + }, + "bench:cases": { + "script": "bench-network-cases.ts" + }, + "bench:latency-sweep": { + "script": "bench-latency-sweep.ts" + }, + "bench:p2p-split-node": { + "script": "bench-p2p-split-node.ts" + }, + "bench:item1": { + "command": "bash", + "args": ["./bench-run-item1.sh"] + }, + "bench:item1:full": { + "command": "bash", + "args": [ + "-lc", + "BENCH_MD_FILE_COUNT=1500 BENCH_MD_MIN_SIZE_BYTES=1024 BENCH_MD_MAX_SIZE_BYTES=20480 BENCH_BIN_FILE_COUNT=500 BENCH_BIN_SIZE_BYTES=102400 BENCH_COUCHDB_RTT_MS=50 bash ./bench-run-item1.sh" + ] + }, + "test:e2e-couchdb": { + "test": ["test-e2e-two-vaults-couchdb.ts"] + }, + "test:journal-chunk-recovery": { + "test": ["test-journal-chunk-recovery.ts"] + }, + "test:journal-receive-failure": { + "test": ["test-journal-receive-failure.ts"] + }, + "test:e2e-matrix": { + "test": ["test-e2e-two-vaults-matrix.ts"] + }, + "test:e2e-matrix:couchdb-enc0": { + "test": ["test-e2e-two-vaults-matrix.ts"], + "filter": "e2e matrix: COUCHDB-enc0" + }, + "test:e2e-matrix:couchdb-enc1": { + "test": ["test-e2e-two-vaults-matrix.ts"], + "filter": "e2e matrix: COUCHDB-enc1" + }, + "test:e2e-matrix:minio-enc0": { + "test": ["test-e2e-two-vaults-matrix.ts"], + "filter": "e2e matrix: MINIO-enc0" + }, + "test:e2e-matrix:minio-enc1": { + "test": ["test-e2e-two-vaults-matrix.ts"], + "filter": "e2e matrix: MINIO-enc1" + } } diff --git a/test/e2e-cli/test-journal-receive-failure.ts b/test/e2e-cli/test-journal-receive-failure.ts new file mode 100644 index 00000000..c659aab9 --- /dev/null +++ b/test/e2e-cli/test-journal-receive-failure.ts @@ -0,0 +1,369 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, readdir, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { GetObjectCommand, ListObjectsV2Command, PutObjectCommand, S3Client } from "@aws-sdk/client-s3"; +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { runCli, runCliOrFail, sanitiseCatStdout } from "./helpers/cli.ts"; +import { startMinio, stopMinio } from "./helpers/docker.ts"; +import { TempDir } from "./helpers/temp.ts"; +import { applyRemoteSyncSettings, initSettingsFile } from "./helpers/settings.ts"; + +type JsonObject = Record; + +function requireEnv(...keys: string[]): string { + for (const key of keys) { + const value = process.env[key]?.trim(); + if (value) return value; + } + throw new Error(`Required env var is missing: ${keys.join(" or ")}`); +} + +function createJournalReceiveBucketName(baseBucket: string, suffix: string): string { + const bucketSuffix = `journal-receive-${suffix}`; + const maxBaseLength = 63 - bucketSuffix.length - 1; + const truncatedBase = baseBucket.slice(0, maxBaseLength).replace(/-+$/, ""); + return truncatedBase ? `${truncatedBase}-${bucketSuffix}` : bucketSuffix; +} + +function isJsonObject(value: unknown): value is JsonObject { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +async function listJournalKeys(client: S3Client, bucket: string): Promise { + const result = await client.send(new ListObjectsV2Command({ Bucket: bucket })); + assert.equal(result.IsTruncated, false, "The disposable test bucket should fit in one S3 listing page"); + return (result.Contents ?? []) + .flatMap(({ Key }) => (Key && /^\d+-docs\.jsonl\.gz$/u.test(Key) ? [Key] : [])) + .sort((left, right) => left.localeCompare(right, undefined, { numeric: true })); +} + +async function addedJournalKey(client: S3Client, bucket: string, before: Set): Promise { + const added = (await listJournalKeys(client, bucket)).filter((key) => !before.has(key)); + assert.equal(added.length, 1, `Expected one new Journal object, got: ${added.join(", ")}`); + const [key] = added; + assert.ok(key); + return key; +} + +async function withCorruptedJournals( + client: S3Client, + bucket: string, + keys: string[], + action: () => Promise +): Promise { + const originals = await Promise.all( + keys.map(async (key) => { + const response = await client.send(new GetObjectCommand({ Bucket: bucket, Key: key })); + const body = response.Body; + assert.ok(body, `Could not read the original Journal bytes for ${key}`); + const bytes = await body.transformToByteArray(); + assert.ok(bytes.length > 0, `The owned Journal object ${key} was empty`); + return { key, bytes }; + }) + ); + + try { + await Promise.all( + originals.map(({ key }) => + client.send( + new PutObjectCommand({ + Bucket: bucket, + Key: key, + Body: Uint8Array.of(0xff, 0xff, 0xff), + }) + ) + ) + ); + await action(); + } finally { + await Promise.all( + originals.map(({ key, bytes }) => + client.send(new PutObjectCommand({ Bucket: bucket, Key: key, Body: bytes })) + ) + ); + } +} + +async function readReceivedJournalKeys(databasePath: string): Promise> { + const scopeRoot = join(databasePath, ".livesync", "runtime", "scopes"); + const files = await readdir(scopeRoot, { recursive: true }); + for (const relativePath of files.filter((file) => file.endsWith("keyvalue-db.json"))) { + const stored: unknown = JSON.parse(await readFile(join(scopeRoot, relativePath), "utf8")); + if (!isJsonObject(stored)) continue; + const entry = Object.entries(stored).find(([key]) => key.includes("bucketsync-checkpoint-")); + if (!entry || !isJsonObject(entry[1])) continue; + const receivedFiles = entry[1]["receivedFiles"]; + if (!isJsonObject(receivedFiles)) continue; + const values = receivedFiles["values"]; + if (receivedFiles["__nodeKvType"] !== "Set" || !Array.isArray(values)) continue; + return new Set(values.filter((value): value is string => typeof value === "string")); + } + throw new Error(`Journal receive checkpoint was not found under ${scopeRoot}`); +} + +async function runScenario(encrypt: boolean): Promise { + const suffix = `${encrypt ? "enc1" : "enc0"}-${Date.now()}-${randomUUID().replaceAll("-", "").slice(0, 8)}`; + const minioEndpoint = requireEnv("MINIO_ENDPOINT", "minioEndpoint").replace(/\/$/, ""); + const minioAccessKey = requireEnv("MINIO_ACCESS_KEY", "accessKey"); + const minioSecretKey = requireEnv("MINIO_SECRET_KEY", "secretKey"); + const minioBucket = createJournalReceiveBucketName(requireEnv("MINIO_BUCKET_NAME", "bucketName"), suffix); + const passphrase = "journal-receive-e2e-passphrase"; + const keepDocker = process.env["LIVESYNC_DEBUG_KEEP_DOCKER"] === "1"; + const stablePath = "e2e/stable.md"; + const middlePath = "e2e/middle.md"; + const secondBrokenPath = "e2e/second-broken.md"; + const laterPath = "e2e/later.md"; + const stableV1 = "Stable file revision one.\n"; + const middleContent = "Content in the intermediate Journal.\n"; + const secondBrokenContent = "Content in the next intermediate Journal.\n"; + const stableV2 = "Stable file revision two.\n"; + const laterContent = "Later Journal content.\n"; + + await using workDir = await TempDir.create(`livesync-cli-journal-receive-${suffix}`); + const sourceDatabase = workDir.join("source-db"); + const sourceVault = workDir.join("source-vault"); + const sourceSettings = workDir.join("source-settings.json"); + const defaultDatabase = workDir.join("default-receiver-db"); + const defaultVault = workDir.join("default-receiver-vault"); + const defaultSettings = workDir.join("default-receiver-settings.json"); + const skipDatabase = workDir.join("skip-receiver-db"); + const skipVault = workDir.join("skip-receiver-vault"); + const skipSettings = workDir.join("skip-receiver-settings.json"); + const sourceStableFile = join(sourceVault, stablePath); + const sourceMiddleFile = join(sourceVault, middlePath); + const sourceSecondBrokenFile = join(sourceVault, secondBrokenPath); + const sourceLaterFile = join(sourceVault, laterPath); + const commonSettings = { + remoteType: "MINIO" as const, + minioBucket, + minioEndpoint, + minioAccessKey, + minioSecretKey, + encrypt, + passphrase, + enableCompression: false, + }; + const client = new S3Client({ + endpoint: minioEndpoint, + region: "us-east-1", + forcePathStyle: true, + credentials: { accessKeyId: minioAccessKey, secretAccessKey: minioSecretKey }, + }); + + await Promise.all([ + mkdir(sourceDatabase, { recursive: true }), + mkdir(sourceVault, { recursive: true }), + mkdir(defaultDatabase, { recursive: true }), + mkdir(defaultVault, { recursive: true }), + mkdir(skipDatabase, { recursive: true }), + mkdir(skipVault, { recursive: true }), + ]); + await mkdir(dirname(sourceStableFile), { recursive: true }); + await writeFile(sourceStableFile, stableV1); + + try { + await startMinio(minioEndpoint, minioAccessKey, minioSecretKey, minioBucket); + await Promise.all([ + initSettingsFile(sourceSettings), + initSettingsFile(defaultSettings), + initSettingsFile(skipSettings), + ]); + await Promise.all([ + applyRemoteSyncSettings(sourceSettings, commonSettings), + applyRemoteSyncSettings(defaultSettings, commonSettings), + applyRemoteSyncSettings(skipSettings, commonSettings), + ]); + const runSource = (...args: string[]) => + runCliOrFail(sourceDatabase, "--settings", sourceSettings, "--vault", sourceVault, ...args); + const runDefault = (...args: string[]) => + runCliOrFail(defaultDatabase, "--settings", defaultSettings, "--vault", defaultVault, ...args); + const tryDefault = (...args: string[]) => + runCli(defaultDatabase, "--settings", defaultSettings, "--vault", defaultVault, ...args); + const runSkip = (...args: string[]) => + runCliOrFail(skipDatabase, "--settings", skipSettings, "--vault", skipVault, ...args); + const trySkip = (...args: string[]) => + runCli(skipDatabase, "--settings", skipSettings, "--vault", skipVault, ...args); + + const beforeInitial = new Set(await listJournalKeys(client, minioBucket)); + await runSource("rebuild", "--force", "--write-settings"); + const firstJournalKey = await addedJournalKey(client, minioBucket, beforeInitial); + await runDefault("fetch", "--force", "--write-settings"); + await runSkip("fetch", "--force", "--write-settings"); + const firstInfo = JSON.parse(await runSource("info", stablePath)) as JsonObject; + assert.equal(typeof firstInfo["revision"], "string"); + const firstRevision = firstInfo["revision"] as string; + for (const receiver of [ + { name: "default receiver", database: defaultDatabase, run: runDefault }, + { name: "skip receiver", database: skipDatabase, run: runSkip }, + ]) { + assert.equal( + sanitiseCatStdout(await receiver.run("cat", stablePath)), + stableV1, + `${receiver.name} should fetch the initial healthy Journal content` + ); + const checkpoint = await readReceivedJournalKeys(receiver.database); + assert.ok( + checkpoint.has(firstJournalKey), + `${receiver.name} should checkpoint the initial healthy Journal` + ); + const info = JSON.parse(await receiver.run("info", stablePath)) as JsonObject; + assert.equal(info["revision"], firstRevision, `${receiver.name} should retain the first source revision`); + assert.equal( + sanitiseCatStdout(await receiver.run("cat-rev", stablePath, firstRevision)), + stableV1, + `${receiver.name} should retain the first revision contents` + ); + } + + await mkdir(dirname(sourceMiddleFile), { recursive: true }); + await writeFile(sourceMiddleFile, middleContent); + await runSource("push", sourceMiddleFile, middlePath); + const beforeMiddle = new Set(await listJournalKeys(client, minioBucket)); + await runSource("sync"); + const brokenJournalKey = await addedJournalKey(client, minioBucket, beforeMiddle); + + await mkdir(dirname(sourceSecondBrokenFile), { recursive: true }); + await writeFile(sourceSecondBrokenFile, secondBrokenContent); + await runSource("push", sourceSecondBrokenFile, secondBrokenPath); + const beforeSecondBroken = new Set(await listJournalKeys(client, minioBucket)); + await runSource("sync"); + const secondBrokenJournalKey = await addedJournalKey(client, minioBucket, beforeSecondBroken); + + await writeFile(sourceStableFile, stableV2); + await writeFile(sourceLaterFile, laterContent); + await runSource("push", sourceStableFile, stablePath); + await runSource("push", sourceLaterFile, laterPath); + const beforeLater = new Set(await listJournalKeys(client, minioBucket)); + await runSource("sync"); + const laterJournalKey = await addedJournalKey(client, minioBucket, beforeLater); + assert.ok(firstJournalKey.localeCompare(brokenJournalKey, undefined, { numeric: true }) < 0); + assert.ok(brokenJournalKey.localeCompare(secondBrokenJournalKey, undefined, { numeric: true }) < 0); + assert.ok(secondBrokenJournalKey.localeCompare(laterJournalKey, undefined, { numeric: true }) < 0); + + const finalInfo = JSON.parse(await runSource("info", stablePath)) as JsonObject; + assert.equal(typeof finalInfo["revision"], "string"); + const finalRevision = finalInfo["revision"] as string; + assert.notEqual(finalRevision, firstRevision); + + await withCorruptedJournals(client, minioBucket, [brokenJournalKey, secondBrokenJournalKey], async () => { + const failedSync = await tryDefault("--debug", "sync"); + assert.notEqual(failedSync.code, 0, "Default receive must report failure for a corrupt Journal"); + assert.match(failedSync.combined, /receive journal error|journal.*failed|replication.*failed/i); + + const checkpoint = await readReceivedJournalKeys(defaultDatabase); + assert.ok(checkpoint.has(firstJournalKey), "Completed earlier Journal must remain checkpointed"); + assert.equal(checkpoint.has(brokenJournalKey), false, "Failed Journal must not be checkpointed"); + assert.equal( + checkpoint.has(secondBrokenJournalKey), + false, + "A later failed Journal must not be checkpointed" + ); + assert.equal(checkpoint.has(laterJournalKey), false, "Failure must prevent checkpointing a later Journal"); + assert.equal( + sanitiseCatStdout(await runDefault("cat", stablePath)), + stableV1, + "The earlier Journal content must remain available after the later failure" + ); + assert.notEqual( + (await tryDefault("cat", middlePath)).code, + 0, + "The broken Journal content must not be applied" + ); + assert.notEqual( + (await tryDefault("cat", secondBrokenPath)).code, + 0, + "The next broken Journal content must not be applied" + ); + assert.notEqual( + (await tryDefault("cat", laterPath)).code, + 0, + "A later Journal must not pass the failed Journal" + ); + }); + + await runDefault("sync"); + assert.equal(sanitiseCatStdout(await runDefault("cat", middlePath)), middleContent); + assert.equal(sanitiseCatStdout(await runDefault("cat", secondBrokenPath)), secondBrokenContent); + assert.equal(sanitiseCatStdout(await runDefault("cat", stablePath)), stableV2); + assert.equal(sanitiseCatStdout(await runDefault("cat", laterPath)), laterContent); + const retriedCheckpoint = await readReceivedJournalKeys(defaultDatabase); + for (const key of [firstJournalKey, brokenJournalKey, secondBrokenJournalKey, laterJournalKey]) { + assert.ok(retriedCheckpoint.has(key), `Retry should checkpoint Journal ${key}`); + } + const retriedInfo = JSON.parse(await runDefault("info", stablePath)) as JsonObject; + assert.equal(retriedInfo["revision"], finalRevision, "Retry should preserve the source stable-file revision"); + assert.equal( + sanitiseCatStdout(await runDefault("cat-rev", stablePath, firstRevision)), + stableV1, + "Retry should retain the earlier stable-file revision" + ); + + await withCorruptedJournals(client, minioBucket, [brokenJournalKey, secondBrokenJournalKey], async () => { + const firstSkip = await trySkip("journal-skip", brokenJournalKey); + assert.notEqual( + firstSkip.code, + 0, + `Skipping the first Journal should stop at the next broken Journal: ${firstSkip.combined}` + ); + assert.ok(firstSkip.combined.includes(brokenJournalKey), "The first skip must identify its Journal key"); + assert.ok( + firstSkip.combined.includes(secondBrokenJournalKey), + "Resumed replication must stop and identify the next broken Journal" + ); + assert.match( + firstSkip.combined, + /decrypt|inflate|decompress|decode|parse|persist|apply/i, + "The first skip attempt must report the failed Journal processing stage" + ); + + const firstSkipCheckpoint = await readReceivedJournalKeys(skipDatabase); + assert.ok(firstSkipCheckpoint.has(firstJournalKey)); + assert.ok(firstSkipCheckpoint.has(brokenJournalKey), "The requested Journal should be skipped"); + assert.equal( + firstSkipCheckpoint.has(secondBrokenJournalKey), + false, + "The first skip must not acknowledge the next failed Journal" + ); + assert.equal(firstSkipCheckpoint.has(laterJournalKey), false); + assert.equal(sanitiseCatStdout(await runSkip("cat", stablePath)), stableV1); + assert.notEqual((await trySkip("cat", middlePath)).code, 0); + assert.notEqual((await trySkip("cat", secondBrokenPath)).code, 0); + assert.notEqual((await trySkip("cat", laterPath)).code, 0); + + const secondSkip = await trySkip("journal-skip", secondBrokenJournalKey); + assert.equal(secondSkip.code, 0, `Explicit second skip should resume replication: ${secondSkip.combined}`); + assert.ok(secondSkip.combined.includes(secondBrokenJournalKey), "The second skip must identify its key"); + assert.match( + secondSkip.combined, + /decrypt|inflate|decompress|decode|parse|persist|apply/i, + "The second skip attempt must report the failed Journal processing stage" + ); + const checkpoint = await readReceivedJournalKeys(skipDatabase); + for (const key of [firstJournalKey, brokenJournalKey, secondBrokenJournalKey, laterJournalKey]) { + assert.ok(checkpoint.has(key), `Explicit skips should advance past Journal ${key}`); + } + assert.equal(sanitiseCatStdout(await runSkip("cat", stablePath)), stableV2); + assert.equal(sanitiseCatStdout(await runSkip("cat", laterPath)), laterContent); + assert.notEqual((await trySkip("cat", middlePath)).code, 0); + assert.notEqual((await trySkip("cat", secondBrokenPath)).code, 0); + const skippedInfo = JSON.parse(await runSkip("info", stablePath)) as JsonObject; + assert.equal(skippedInfo["revision"], finalRevision, "Later valid data should retain its source revision"); + assert.equal( + sanitiseCatStdout(await runSkip("cat-rev", stablePath, firstRevision)), + stableV1, + "Later valid data should retain earlier revision history" + ); + }); + } finally { + client.destroy(); + if (!keepDocker) { + await stopMinio().catch(() => {}); + } + } +} + +test("journal receive failure, retry, and targeted skips through MINIO", async (t) => { + await t.test("without encryption", async () => runScenario(false)); + await t.test("with encryption", async () => runScenario(true)); +}); diff --git a/test/e2e-obsidian/scripts/journal-nfc-resend.ts b/test/e2e-obsidian/scripts/journal-nfc-resend.ts index 40fffc1e..0618c0e6 100644 --- a/test/e2e-obsidian/scripts/journal-nfc-resend.ts +++ b/test/e2e-obsidian/scripts/journal-nfc-resend.ts @@ -2,6 +2,7 @@ import { randomBytes } from "node:crypto"; import { createServer } from "node:net"; import { readFile, readdir } from "node:fs/promises"; import { join } from "node:path"; +import { PutObjectCommand } from "@aws-sdk/client-s3"; import { deriveIdKey } from "@vrtmrz/livesync-commonlib/settings"; import { evalObsidianJson } from "../runner/cli.ts"; import { discoverObsidianCli, requireObsidianBinary } from "../runner/environment.ts"; @@ -17,6 +18,8 @@ import { type LocalDatabaseEntry, } from "../runner/liveSyncWorkflow.ts"; import { + createObjectStorageClient, + readObjectStorageObject, deleteObjectStoragePrefix, ensureObjectStorageBucket, listObjectStorageObjects, @@ -27,7 +30,7 @@ import { } from "../runner/objectStorage.ts"; import { startObsidianLiveSyncSession, type ObsidianLiveSyncSession } from "../runner/session.ts"; import { createTemporaryVault, type TemporaryVault } from "../runner/vault.ts"; -import { withObsidianPage } from "../runner/ui.ts"; +import { openLiveSyncSettings, waitForVisibleObsidianDialogue, withObsidianPage } from "../runner/ui.ts"; process.env.E2E_OBSIDIAN_CLI_TIMEOUT_MS ??= "60000"; @@ -557,6 +560,141 @@ async function waitForJournalObjects( ); } +async function verifySingleJournalRecovery( + cliBinary: string, + session: ObsidianLiveSyncSession, + vaultPath: string, + storage: ObjectStorageConfig, + prefix: string, + validJournalKey: string +): Promise { + const objects = await listObjectStorageObjects(storage, prefix); + const timestamp = + Math.max(Date.now(), ...objects.map(({ Key }) => Number(Key?.slice(prefix.length).split("-")[0]) || 0)) + 1; + const firstKey = `${timestamp}-docs.jsonl.gz`; + const secondKey = `${timestamp + 1}-docs.jsonl.gz`; + const laterKey = `${timestamp + 2}-docs.jsonl.gz`; + const validBytes = await readObjectStorageObject(storage, validJournalKey); + const client = createObjectStorageClient(storage); + try { + for (const [key, bytes] of [ + [firstKey, new Uint8Array([1])], + [secondKey, new Uint8Array([1])], + [laterKey, validBytes], + ] as const) { + await client.send(new PutObjectCommand({ Bucket: storage.bucket, Key: prefix + key, Body: bytes })); + } + } finally { + client.destroy(); + } + + type RecoveryState = { key: string | null; stage: string | null; received: string[] }; + const readState = async (): Promise => + await evalObsidianJson( + cliBinary, + [ + "(async()=>{", + "const core=app.plugins.plugins['obsidian-livesync'].core;", + "const replicator=core.services.replicator.getActiveReplicator();", + "const stopped=replicator.getStoppedJournal();", + "const checkpoint=await replicator.client.getCheckpointInfo();", + "return JSON.stringify({key:stopped?.key??null,stage:stopped?.stage??null,received:[...checkpoint.receivedFiles]});", + "})()", + ].join(""), + session.cliEnv, + timeoutMs + ); + const waitForState = async (expectedKey: string | null, expectedReceived: string[]): Promise => { + const deadline = Date.now() + timeoutMs; + let state: RecoveryState | undefined; + while (Date.now() < deadline) { + state = await readState(); + if (state.key === expectedKey && expectedReceived.every((key) => state?.received.includes(key))) return; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`Journal recovery did not reach the expected state: ${JSON.stringify(state)}`); + }; + const result = await evalObsidianJson( + cliBinary, + [ + "(async()=>{", + "const core=app.plugins.plugins['obsidian-livesync'].core;", + "const replicator=core.services.replicator.getActiveReplicator();", + "return JSON.stringify(await replicator.replicateAllFromServer(core.settings,true));", + "})()", + ].join(""), + session.cliEnv, + timeoutMs + ); + assertEqual(result, false, "Receipt must stop at the first malformed Journal."); + const initial = await readState(); + assertEqual(initial.key, firstKey, "The first failed Journal must be offered for recovery."); + for (const key of [firstKey, secondKey, laterKey]) { + assertEqual(initial.received.includes(key), false, "Failed and later Journals must remain unacknowledged."); + } + + await withObsidianPage(session.remoteDebuggingPort, async (page) => { + const navigator = await openLiveSyncSettings(page, uiTimeoutMs); + try { + const maintenance = await navigator.openPage("Maintenance"); + assertEqual( + await maintenance.getByText("Ignore broken journals (risk of data loss)", { exact: true }).count(), + 0, + "The persistent skip toggle must not be offered." + ); + const setting = maintenance.locator(".setting-item").filter({ + has: navigator.page.getByText("Skip stopped Journal and resume", { exact: true }), + }); + await setting.waitFor({ state: "visible", timeout: uiTimeoutMs }); + const choose = async (key: string, accept: boolean): Promise => { + await setting + .getByRole("button", { name: "Skip and resume", exact: true }) + .click({ timeout: uiTimeoutMs }); + const dialogue = await waitForVisibleObsidianDialogue(page, key, uiTimeoutMs); + const text = await dialogue.innerText(); + if (!text.includes(key) || !/missing/i.test(text)) { + throw new Error( + "The skip confirmation must identify the exact Journal and its missing-content risk." + ); + } + await dialogue + .getByRole("button", { name: accept ? "Yes" : "No", exact: true }) + .click({ timeout: uiTimeoutMs }); + }; + await choose(firstKey, false); + assertEqual( + JSON.stringify(await readState()), + JSON.stringify(initial), + "Cancelling must preserve the failure and checkpoint." + ); + await choose(firstKey, true); + await waitForState(secondKey, [firstKey]); + const afterFirst = await readState(); + assertEqual( + afterFirst.received.includes(secondKey), + false, + "One action must not skip the next failed Journal." + ); + assertEqual( + afterFirst.received.includes(laterKey), + false, + "Receipt must stop before the later valid Journal." + ); + await choose(secondKey, true); + await waitForState(null, [firstKey, secondKey, laterKey]); + } finally { + await navigator.close(); + } + }); + const saved = JSON.parse( + await readFile(join(vaultPath, ".obsidian/plugins/obsidian-livesync/data.json"), "utf8") + ) as Record; + assertEqual("ignoreBrokenJournals" in saved, false, "Recovery must not persist a skip toggle."); + console.log( + "Journal recovery UI: cancellation preserved the checkpoint, each action skipped one named Journal, and the next failure stopped receipt." + ); +} + async function postponeConflictIfOpen(session: ObsidianLiveSyncSession): Promise { await withObsidianPage(session.remoteDebuggingPort, async (page) => { const dialogue = page.locator(".modal-container:visible").filter({ hasText: "Conflicting changes" }); @@ -794,6 +932,24 @@ async function runCase( ); await waitForSinglePhysicalUnicodeFile(receiverVault.path, controlPathNfd, controlContent); + const validJournalKey = rebuiltKeys.find((key) => key.endsWith("-docs.jsonl.gz")); + if (!validJournalKey) throw new Error("A valid Journal fixture is required for recovery testing."); + await verifySingleJournalRecovery( + cliBinary, + receiverSession, + receiverVault.path, + objectStorage, + bucketPrefix, + validJournalKey + ); + await postponeConflictIfOpen(receiverSession); + assertSameConflictTree( + await readConflictTree(cliBinary, receiverSession.cliEnv, conflictPathNfc, initialEntry.rev, false), + sourceTree, + "Receiver after explicit Journal skips" + ); + await waitForSinglePhysicalUnicodeFile(receiverVault.path, controlPathNfd, controlContent); + const initialJournalObjects = initialKeys.filter((key) => key.endsWith("-docs.jsonl.gz")).length; const rebuiltJournalObjects = rebuiltKeys.filter((key) => key.endsWith("-docs.jsonl.gz")).length; console.log( diff --git a/updates.md b/updates.md index b44a0cea..0fe8ecf7 100644 --- a/updates.md +++ b/updates.md @@ -18,6 +18,12 @@ Earlier releases remain available in the 1.0 release history, the 1.0 preview hi ### Fixed +#### Object Storage + +- Failed Journal writes now stop replication instead of silently passing over missing records. After correcting the cause, the next replication with the same local database retries the failed Journal. + - If a Journal cannot be recovered and missing content or history is acceptable, **Maintenance → Syncing → Skip stopped Journal and resume** asks for confirmation of its exact key and failure stage, then skips only that Journal and resumes receipt. The next Journal failure still stops replication. The action does not add a persistent ignore setting. + - **Reset journal received history** replays Journals which remain available remotely, except those sent by this device. It cannot recreate missing remote Journals; earlier failures already silently acknowledged also require this explicit replay. + #### Conflict resolution - Disabling or reloading the plug-in no longer causes a secondary error when pending work writes to the log. Logs arriving after the log display closes go to Obsidian's console. @@ -36,6 +42,8 @@ Earlier releases remain available in the 1.0 release history, the 1.0 preview hi #### Fixed +- Commonlib warnings now appear on stderr without `--debug`, so deliberately skipped Journals and other important notices remain visible during ordinary CLI use. + - NFC file lookups and directory scans report permission and I/O failures instead of treating unreadable files as absent or returning a successful partial scan. A scan retains the comparison settings with which it began. ## 1.0.44 From 3a7f3029a3ac5a587330d11111892f220cedc771 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 5 Nov 2026 02:44:18 +0000 Subject: [PATCH 2/5] Extract the Journal skip command and clarify manual recovery --- src/apps/cli/commands/journalSkip.ts | 74 ++++++ .../cli/commands/journalSkip.unit.spec.ts | 213 ++++++++++++++++++ src/apps/cli/commands/runCommand.ts | 62 +---- src/apps/cli/package.json | 2 +- .../SettingDialogue/PaneMaintenance.ts | 7 +- .../PaneMaintenance.unit.spec.ts | 10 +- 6 files changed, 307 insertions(+), 61 deletions(-) create mode 100644 src/apps/cli/commands/journalSkip.ts create mode 100644 src/apps/cli/commands/journalSkip.unit.spec.ts 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 = ""; From 9e564eb51e4e768a8466161aa73ba131a1bf9a1c Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 5 Nov 2026 04:47:31 +0000 Subject: [PATCH 3/5] Adopt Commonlib 0.3.2 for stopped-Journal recovery --- docs/design_docs_of_journalsync_2nd.md | 4 +++- package-lock.json | 8 ++++---- package.json | 2 +- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/docs/design_docs_of_journalsync_2nd.md b/docs/design_docs_of_journalsync_2nd.md index c285520a..2a5af108 100644 --- a/docs/design_docs_of_journalsync_2nd.md +++ b/docs/design_docs_of_journalsync_2nd.md @@ -1,7 +1,9 @@ ## The design document of the Journal Replicator 2nd Edition Implementation note: the receive-failure correction and single-Journal skip -action are unreleased. Their owning contract is Commonlib's +API are published in [Commonlib 0.3.2](https://github.com/vrtmrz/livesync-commonlib/releases/tag/0.3.2). +The Self-hosted LiveSync host action remains unreleased. The owning contract +is Commonlib's [Journal receive failures and recovery](https://github.com/vrtmrz/livesync-commonlib/blob/main/docs/journal-receive-recovery.md). The receiver acknowledges successful processing, stops on failure by default, and retries from that Journal with the same database. Explicit skipping does diff --git a/package-lock.json b/package-lock.json index 9a6a2ba3..f3d42048 100644 --- a/package-lock.json +++ b/package-lock.json @@ -23,7 +23,7 @@ "@smithy/types": "^4.14.3", "@smithy/util-retry": "^4.4.5", "@vrtmrz/browser-ui-kit": "0.1.0", - "@vrtmrz/livesync-commonlib": "0.3.1", + "@vrtmrz/livesync-commonlib": "0.3.2", "@vrtmrz/obsidian-plugin-kit": "0.1.7", "@vrtmrz/ui-interactions": "0.1.2", "@vrtmrz/value-store": "0.1.0", @@ -4065,9 +4065,9 @@ } }, "node_modules/@vrtmrz/livesync-commonlib": { - "version": "0.3.1", - "resolved": "https://registry.npmjs.org/@vrtmrz/livesync-commonlib/-/livesync-commonlib-0.3.1.tgz", - "integrity": "sha512-/vm/5MEiFIS1mEfW/+kPaqbx/cpgXlEl+uQHTm5ClHxNZ/umARuxXZX33TlxFpVimLVjYPRQ97XSZAz5HKy8vw==", + "version": "0.3.2", + "resolved": "https://registry.npmjs.org/@vrtmrz/livesync-commonlib/-/livesync-commonlib-0.3.2.tgz", + "integrity": "sha512-PiKUdJUiBI4ot0AV1BjlkDoaY4Lbn9oTdEjL8FMX1qf2VCOwSnbsLoQMUr/45wmINpV31KWmqgr2zxqcYO+hpA==", "license": "MIT", "dependencies": { "@aws-crypto/sha256-js": "5.2.0", diff --git a/package.json b/package.json index 9f81a6ef..20c6efbc 100644 --- a/package.json +++ b/package.json @@ -211,7 +211,7 @@ "@smithy/types": "^4.14.3", "@smithy/util-retry": "^4.4.5", "@vrtmrz/browser-ui-kit": "0.1.0", - "@vrtmrz/livesync-commonlib": "0.3.1", + "@vrtmrz/livesync-commonlib": "0.3.2", "@vrtmrz/obsidian-plugin-kit": "0.1.7", "@vrtmrz/ui-interactions": "0.1.2", "@vrtmrz/value-store": "0.1.0", From a0f14838913d0079c7eaaa08510dd21758f0fdaa Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 5 Nov 2026 08:36:36 +0000 Subject: [PATCH 4/5] Route Journal recovery notices through the existing logger --- .../SettingDialogue/PaneMaintenance.ts | 21 +++++++++++-------- .../PaneMaintenance.unit.spec.ts | 18 ++++++++++------ 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.ts b/src/modules/features/SettingDialogue/PaneMaintenance.ts index b2384dbf..d32ad38b 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.ts +++ b/src/modules/features/SettingDialogue/PaneMaintenance.ts @@ -1,6 +1,5 @@ import { EVENT_REQUEST_PERFORM_GC_V3, eventHub } from "@/common/events.ts"; import { $msg } from "@/common/translation"; -import { Notice } from "@/deps.ts"; import { createCoreSettingsAfterFullReset, createEditingSettingsAfterFullReset, @@ -171,19 +170,21 @@ export function paneMaintenance( .onClick(async () => { const replicator = this.core.replicator; if (!isLiveSyncJournalReplicator(replicator)) { - new Notice( + Logger( $msg( "No stopped Journal is available. Retry replication to identify the stopped Journal." - ) + ), + LOG_LEVEL_NOTICE ); return; } const stoppedJournal: Readonly | undefined = replicator.getStoppedJournal(); if (!stoppedJournal) { - new Notice( + Logger( $msg( "No stopped Journal is available. Retry replication to identify the stopped Journal." - ) + ), + LOG_LEVEL_NOTICE ); return; } @@ -202,10 +203,11 @@ export function paneMaintenance( return; } if (this.core.replicator !== replicator) { - new Notice( + Logger( $msg( "The active replicator changed. Retry replication to identify the stopped Journal." - ) + ), + LOG_LEVEL_NOTICE ); return; } @@ -214,10 +216,11 @@ export function paneMaintenance( stoppedJournal ); if (!resumed) { - new Notice( + Logger( $msg( "Journal receipt did not complete after this action. Check the current failure before retrying." - ) + ), + LOG_LEVEL_NOTICE ); } }) diff --git a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts b/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts index f96bbb62..3df3acb8 100644 --- a/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts +++ b/src/modules/features/SettingDialogue/PaneMaintenance.unit.spec.ts @@ -326,9 +326,11 @@ describe("paneMaintenance", () => { expect(getStoppedJournal).toHaveBeenCalledOnce(); expect(rendered.askYesNoDialog).not.toHaveBeenCalled(); expect(skipStoppedJournalAndResume).not.toHaveBeenCalled(); - expect(maintenanceHarness.notice).toHaveBeenCalledWith( - "No stopped Journal is available. Retry replication to identify the stopped Journal." + expect(maintenanceHarness.logger).toHaveBeenCalledWith( + "No stopped Journal is available. Retry replication to identify the stopped Journal.", + "notice" ); + expect(maintenanceHarness.notice).not.toHaveBeenCalled(); }); it("does not skip the stopped Journal when confirmation is cancelled", async () => { @@ -378,9 +380,11 @@ describe("paneMaintenance", () => { expect(active.skipStoppedJournalAndResume).not.toHaveBeenCalled(); expect(replacement.skipStoppedJournalAndResume).not.toHaveBeenCalled(); - expect(maintenanceHarness.notice).toHaveBeenCalledWith( - "The active replicator changed. Retry replication to identify the stopped Journal." + expect(maintenanceHarness.logger).toHaveBeenCalledWith( + "The active replicator changed. Retry replication to identify the stopped Journal.", + "notice" ); + expect(maintenanceHarness.notice).not.toHaveBeenCalled(); }); it("reports when receipt does not complete after skip and resume", async () => { @@ -391,8 +395,10 @@ describe("paneMaintenance", () => { await rendered.click(); expect(skipStoppedJournalAndResume).toHaveBeenCalledOnce(); - expect(maintenanceHarness.notice).toHaveBeenCalledWith( - "Journal receipt did not complete after this action. Check the current failure before retrying." + expect(maintenanceHarness.logger).toHaveBeenCalledWith( + "Journal receipt did not complete after this action. Check the current failure before retrying.", + "notice" ); + expect(maintenanceHarness.notice).not.toHaveBeenCalled(); }); }); From e7124eda984039ed711c776f8b75bc5f0659be06 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Thu, 5 Nov 2026 08:36:36 +0000 Subject: [PATCH 5/5] Keep Commonlib CLI diagnostics behind debug mode --- src/apps/cli/README.md | 5 +++++ src/apps/cli/main.log-handler.unit.spec.ts | 11 ++-------- src/apps/cli/main.ts | 13 +++++++---- test/e2e-cli/test-journal-receive-failure.ts | 23 ++++++++++---------- 4 files changed, 28 insertions(+), 24 deletions(-) diff --git a/src/apps/cli/README.md b/src/apps/cli/README.md index 99614627..a74107e8 100644 --- a/src/apps/cli/README.md +++ b/src/apps/cli/README.md @@ -116,6 +116,11 @@ saved. After correcting the cause, run `sync` again with the same database and settings to retry it. Existing partial writes are retained. `fetch --force` replaces the local database instead and follows the reconstruction rules below. +A failed `sync` prints `[Error] Command 'sync' failed` and exits with code `1`. +To inspect the Journal key, failure stage, and diagnostic details, retry with +`--debug`, for example `livesync-cli ./my-db --debug sync`, retaining the same +database and settings. Debug diagnostics are written to standard error. + If the Journal cannot be recovered and missing content or history is acceptable, run `journal-skip `. The CLI first performs normal replication with the retained local database. It skips only when that attempt diff --git a/src/apps/cli/main.log-handler.unit.spec.ts b/src/apps/cli/main.log-handler.unit.spec.ts index 5c344a10..20b8037c 100644 --- a/src/apps/cli/main.log-handler.unit.spec.ts +++ b/src/apps/cli/main.log-handler.unit.spec.ts @@ -79,15 +79,6 @@ describe("CLI log handler", () => { it.each([ ["NOTICE", LOG_LEVEL_NOTICE], ["URGENT", LOG_LEVEL_URGENT], - ] as const)("writes Commonlib %s diagnostics to stderr by default", async (name, level) => { - const message = `Commonlib ${name} diagnostic`; - const standardIo = await invokeGlobalLogger(databasePath, message, level); - - expect(standardIo.writeStderr).toHaveBeenCalledWith(expect.stringContaining(`[${level}] ${message}\n`)); - expect(standardIo.writeStdout).not.toHaveBeenCalledWith(expect.stringContaining(message)); - }); - - it.each([ ["DEBUG", LOG_LEVEL_DEBUG], ["INFO", LOG_LEVEL_INFO], ["VERBOSE", LOG_LEVEL_VERBOSE], @@ -101,6 +92,8 @@ describe("CLI log handler", () => { it("writes detailed Commonlib diagnostics to stderr in debug mode", async () => { const messages = [ + ["NOTICE", LOG_LEVEL_NOTICE], + ["URGENT", LOG_LEVEL_URGENT], ["DEBUG", LOG_LEVEL_DEBUG], ["INFO", LOG_LEVEL_INFO], ["VERBOSE", LOG_LEVEL_VERBOSE], diff --git a/src/apps/cli/main.ts b/src/apps/cli/main.ts index 511e015d..6c93dd93 100644 --- a/src/apps/cli/main.ts +++ b/src/apps/cli/main.ts @@ -378,10 +378,15 @@ export async function main( const writeLine = avoidStdoutNoise ? writeStderrLine : writeStdoutLine; writeLine(standardIo, ...values); }; - setGlobalLogFunction((msg, level) => { - if (!options.debug && level !== LOG_LEVEL_NOTICE && level !== LOG_LEVEL_URGENT) return; - writeStderrLine(standardIo, `[${level}]`, msg); - }); + if (options.debug) { + setGlobalLogFunction((msg, level) => { + writeStderrLine(standardIo, `[${level}]`, msg); + }); + } else { + setGlobalLogFunction((msg, level) => { + // NO OP, leave it to logFunction + }); + } if (options.command === "init-settings") { await createDefaultSettingsFile(options, standardIo); return; diff --git a/test/e2e-cli/test-journal-receive-failure.ts b/test/e2e-cli/test-journal-receive-failure.ts index c659aab9..cb5f43e2 100644 --- a/test/e2e-cli/test-journal-receive-failure.ts +++ b/test/e2e-cli/test-journal-receive-failure.ts @@ -247,9 +247,12 @@ async function runScenario(encrypt: boolean): Promise { assert.notEqual(finalRevision, firstRevision); await withCorruptedJournals(client, minioBucket, [brokenJournalKey, secondBrokenJournalKey], async () => { - const failedSync = await tryDefault("--debug", "sync"); - assert.notEqual(failedSync.code, 0, "Default receive must report failure for a corrupt Journal"); - assert.match(failedSync.combined, /receive journal error|journal.*failed|replication.*failed/i); + const failedSync = await tryDefault("sync"); + assert.equal(failedSync.code, 1, "Default receive must report failure for a corrupt Journal"); + assert.match(failedSync.stderr, /\[Error\] Command 'sync' failed/); + const debugRetry = await tryDefault("--debug", "sync"); + assert.equal(debugRetry.code, 1, "Debug retry must still report the corrupt Journal"); + assert.ok(debugRetry.stderr.includes(`Receive Journal ${brokenJournalKey} failed at `)); const checkpoint = await readReceivedJournalKeys(defaultDatabase); assert.ok(checkpoint.has(firstJournalKey), "Completed earlier Journal must remain checkpointed"); @@ -307,14 +310,12 @@ async function runScenario(encrypt: boolean): Promise { `Skipping the first Journal should stop at the next broken Journal: ${firstSkip.combined}` ); assert.ok(firstSkip.combined.includes(brokenJournalKey), "The first skip must identify its Journal key"); + assert.match(firstSkip.stderr, /\[Error\] Command 'journal-skip' failed/); + const nextFailure = await trySkip("--debug", "sync"); + assert.equal(nextFailure.code, 1, "Debug retry must stop at the next broken Journal"); assert.ok( - firstSkip.combined.includes(secondBrokenJournalKey), - "Resumed replication must stop and identify the next broken Journal" - ); - assert.match( - firstSkip.combined, - /decrypt|inflate|decompress|decode|parse|persist|apply/i, - "The first skip attempt must report the failed Journal processing stage" + nextFailure.stderr.includes(`Receive Journal ${secondBrokenJournalKey} failed at `), + "Debug retry must identify the next broken Journal and its failure stage" ); const firstSkipCheckpoint = await readReceivedJournalKeys(skipDatabase); @@ -331,7 +332,7 @@ async function runScenario(encrypt: boolean): Promise { assert.notEqual((await trySkip("cat", secondBrokenPath)).code, 0); assert.notEqual((await trySkip("cat", laterPath)).code, 0); - const secondSkip = await trySkip("journal-skip", secondBrokenJournalKey); + const secondSkip = await trySkip("--debug", "journal-skip", secondBrokenJournalKey); assert.equal(secondSkip.code, 0, `Explicit second skip should resume replication: ${secondSkip.combined}`); assert.ok(secondSkip.combined.includes(secondBrokenJournalKey), "The second skip must identify its key"); assert.match(