From a94c3fef90acd1a559c9cf760b5141ed69cd7229 Mon Sep 17 00:00:00 2001 From: vorotamoroz Date: Wed, 4 Nov 2026 15:25:30 +0000 Subject: [PATCH] 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