diff --git a/src/apps/cli/README.md b/src/apps/cli/README.md index 8761bb08..4aff7dad 100644 --- a/src/apps/cli/README.md +++ b/src/apps/cli/README.md @@ -452,6 +452,44 @@ Use this command to select a target for `p2p-sync`. - Stop the host with `Ctrl+C`. - In CLI mode, behaviour is non-interactive and acceptance follows settings. +After the local database is ready, the host restores its configured P2P settings +before connecting. It opens the connection explicitly, including when +`P2P_AutoStart` is `false`. + +- `P2P_AutoSyncPeers` selects peers for an initial bidirectional synchronisation. +- `P2P_AutoWatchPeers` selects peers whose later announced changes the host fetches. + Those peers must enable `P2P_AutoBroadcast`. +- `P2P_AutoBroadcast` enables announcements of changes in the host's local database. + Clients must follow the host to fetch those changes automatically. +- Headless operation requires accepted peers. Configure `P2P_AutoAcceptingPeers` + or retain an existing acceptance decision; selecting a peer for automatic + synchronisation does not itself accept that peer. + +For example, a P2P-only host for peers named `Desktop` and `android` can use: + +```json +{ + "remoteType": "ONLY_P2P", + "P2P_Enabled": true, + "P2P_IsHeadless": true, + "P2P_AutoStart": false, + "P2P_AutoBroadcast": true, + "P2P_AutoAcceptingPeers": "Desktop,android", + "P2P_AutoSyncPeers": "Desktop,android", + "P2P_AutoWatchPeers": "Desktop,android" +} +``` + +Add these fields to settings which already contain the shared P2P room, +passphrase, signalling relay, and this host's unique peer name. Configure the +clients to accept, synchronise with, and follow the host, and to announce their +own changes. With a persistent local database, the host can retain received +changes for a client which connects later. + +`p2p-host` exchanges database content. It does not scan or watch external Vault +file edits. Single-operation CLI commands keep automatic synchronisation +suspended during their execution. + ##### info `info` output fields: diff --git a/src/apps/cli/commands/p2p.ts b/src/apps/cli/commands/p2p.ts index 97bd1602..1cbeff79 100644 --- a/src/apps/cli/commands/p2p.ts +++ b/src/apps/cli/commands/p2p.ts @@ -5,6 +5,7 @@ import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFu import { LiveSyncError } from "@vrtmrz/livesync-commonlib/compat/common/LSError"; import type { P2PPeerConnectionMetrics, P2PServiceViews } from "@vrtmrz/livesync-commonlib/p2p"; import { fsPromises } from "@vrtmrz/livesync-commonlib/node"; +import type { CLIP2PHostSettings } from "./types"; type CLIP2PPeer = { peerId: string; @@ -191,8 +192,11 @@ export async function syncWithPeer( /** Connect the headless P2P host and transfer transport ownership to the caller. */ export async function openP2PHost( core: LiveSyncBaseCore, - p2pService: CLIP2PService | undefined + p2pService: CLIP2PService | undefined, + originalP2PSettings: CLIP2PHostSettings ): Promise { + // Restore only P2P policy after database readiness, before peer discovery. + await core.services.setting.applyPartial(originalP2PSettings); const service = requireP2PService(core, p2pService); await service.transportLifecycle.connect(); return service; diff --git a/src/apps/cli/commands/p2p.unit.spec.ts b/src/apps/cli/commands/p2p.unit.spec.ts index 8e14a99b..6503eb73 100644 --- a/src/apps/cli/commands/p2p.unit.spec.ts +++ b/src/apps/cli/commands/p2p.unit.spec.ts @@ -1,5 +1,7 @@ import { describe, expect, it, vi } from "vitest"; -import { collectPeers, createPeerConnectionStatsPayload, parseTimeoutSeconds, syncWithPeer } from "./p2p"; +import { collectPeers, createPeerConnectionStatsPayload, openP2PHost, parseTimeoutSeconds, syncWithPeer } from "./p2p"; +import { AutoAccepting } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import type { CLIP2PHostSettings } from "./types"; function createCore() { const settings = { P2P_Enabled: true, P2P_AppID: "app-id", P2P_IsHeadless: false }; @@ -37,6 +39,87 @@ function createP2PService() { } describe("p2p command helpers", () => { + const passiveHostSettings: CLIP2PHostSettings = { + P2P_Enabled: true, + P2P_AutoAccepting: AutoAccepting.NONE, + P2P_AutoStart: false, + P2P_AutoBroadcast: false, + P2P_AutoSyncPeers: "", + P2P_AutoWatchPeers: "", + }; + + function createHostCore() { + const settings = { + ...passiveHostSettings, + P2P_Enabled: false, + P2P_IsHeadless: false, + P2P_AppID: "app-id", + liveSync: false, + suspendFileWatching: true, + }; + const applyPartial = vi.fn(async (partial: CLIP2PHostSettings) => { + Object.assign(settings, partial); + }); + const saveSettingData = vi.fn(); + const applySettings = vi.fn(); + return { + core: { + services: { + setting: { currentSettings: () => settings, applyPartial, saveSettingData }, + control: { applySettings }, + }, + }, + settings, + applyPartial, + saveSettingData, + applySettings, + }; + } + + it.each([ + passiveHostSettings, + { ...passiveHostSettings, P2P_AutoBroadcast: true, P2P_AutoWatchPeers: "Desktop" }, + { ...passiveHostSettings, P2P_AutoAccepting: AutoAccepting.ALL, P2P_AutoSyncPeers: "Desktop,android" }, + ])("connects with the configured host policy without saving or restarting the lifecycle (%j)", async (original) => { + const host = createHostCore(); + const { service, connect, disconnect } = createP2PService(); + connect.mockImplementation(async () => { + expect(host.settings).toMatchObject({ ...original, P2P_IsHeadless: true }); + }); + + await expect(openP2PHost(host.core as never, service as never, original)).resolves.toBe(service); + + expect(host.applyPartial).toHaveBeenCalledExactlyOnceWith(original); + expect(connect).toHaveBeenCalledOnce(); + expect(disconnect).not.toHaveBeenCalled(); + expect(host.settings).toMatchObject({ liveSync: false, suspendFileWatching: true }); + expect(host.saveSettingData).not.toHaveBeenCalled(); + expect(host.applySettings).not.toHaveBeenCalled(); + }); + + it("keeps P2P disabled when the saved host policy disables it", async () => { + const host = createHostCore(); + const { service, connect } = createP2PService(); + + await expect( + openP2PHost(host.core as never, service as never, { ...passiveHostSettings, P2P_Enabled: false }) + ).rejects.toThrow("P2P is disabled"); + + expect(connect).not.toHaveBeenCalled(); + }); + + it("does not connect when restoring the host policy fails", async () => { + const host = createHostCore(); + const { service, connect } = createP2PService(); + host.applyPartial.mockRejectedValue(new Error("settings failed")); + + await expect(openP2PHost(host.core as never, service as never, passiveHostSettings)).rejects.toThrow( + "settings failed" + ); + + expect(connect).not.toHaveBeenCalled(); + }); + it("accepts non-negative timeout", () => { expect(parseTimeoutSeconds("0", "p2p-peers")).toBe(0); expect(parseTimeoutSeconds("2.5", "p2p-sync")).toBe(2.5); diff --git a/src/apps/cli/commands/runCommand.ts b/src/apps/cli/commands/runCommand.ts index 8f27b870..26e22f9b 100644 --- a/src/apps/cli/commands/runCommand.ts +++ b/src/apps/cli/commands/runCommand.ts @@ -214,7 +214,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext if (options.command === "p2p-host") { writeStderrLine(standardIo, "[Command] p2p-host"); - await openP2PHost(core, context.p2pReplicator); + await openP2PHost(core, context.p2pReplicator, context.originalP2PSettings); writeStderrLine(standardIo, "[Ready] P2P host is running. Press Ctrl+C to stop."); await new Promise(() => {}); return true; diff --git a/src/apps/cli/commands/types.ts b/src/apps/cli/commands/types.ts index 58576176..a60eeaa5 100644 --- a/src/apps/cli/commands/types.ts +++ b/src/apps/cli/commands/types.ts @@ -49,6 +49,16 @@ export interface CLIOptions { interval?: number; } +export type CLIP2PHostSettings = Pick< + ObsidianLiveSyncSettings, + | "P2P_Enabled" + | "P2P_AutoAccepting" + | "P2P_AutoStart" + | "P2P_AutoBroadcast" + | "P2P_AutoSyncPeers" + | "P2P_AutoWatchPeers" +>; + export interface CLICommandContext { databasePath: string; vaultPath: string; @@ -58,6 +68,8 @@ export interface CLICommandContext { /** Current-result contract owned by the P2P service feature. */ p2pReplicator?: UseP2PReplicatorResult; settingsPath: string; + /** Configured P2P policy before temporary CLI startup suspension. */ + originalP2PSettings: CLIP2PHostSettings; originalSyncSettings: Pick< ObsidianLiveSyncSettings, | "liveSync" diff --git a/src/apps/cli/main.bootstrap.unit.spec.ts b/src/apps/cli/main.bootstrap.unit.spec.ts index 91679968..a63fcbd1 100644 --- a/src/apps/cli/main.bootstrap.unit.spec.ts +++ b/src/apps/cli/main.bootstrap.unit.spec.ts @@ -4,12 +4,20 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import * as chokidar from "chokidar"; import { ControlService } from "@vrtmrz/livesync-commonlib/compat/services/base/ControlService"; -import type { FilePathWithPrefix } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import { + AutoAccepting, + REMOTE_COUCHDB, + REMOTE_P2P, + type FilePathWithPrefix, + type ObsidianLiveSyncSettings, +} from "@vrtmrz/livesync-commonlib/compat/common/types"; import { ServiceFileHandler } from "@/serviceModules/FileHandler"; import { ServiceFileAccessCLI } from "./serviceModules/ServiceFileAccessImpl"; import { runCommand } from "./commands/runCommand"; import { createDefaultCliSettings } from "./cliSettingsDefaults"; import { main, type CliCommandRunner } from "./main"; +import { cloneSettings } from "./settingsPersistence"; +import type { CLICommand } from "./commands/types"; vi.mock("chokidar", { spy: true }); @@ -61,12 +69,107 @@ describe("CLI database preparation", () => { await rm(directory, { recursive: true, force: true }); }); - async function start(command: "daemon" | "mirror" | "ls", runner: CliCommandRunner, exitCode = 1) { + async function start(command: CLICommand, runner: CliCommandRunner, exitCode = 1) { process.argv = ["node", "livesync-cli", directory, "--vault", vaultPath, "--settings", settingsPath, command]; // Daemon probes return false so the real core unloads without keeping a daemon alive. await expect(main(standardIo, runner)).rejects.toThrow(`__EXIT__:${exitCode}`); } + it.each([REMOTE_P2P, REMOTE_COUCHDB])( + "restores configured P2P automation before the host connects (%s)", + async (remoteType) => { + const p2pSettings = { + P2P_Enabled: true, + P2P_AutoAccepting: AutoAccepting.ALL, + P2P_AutoStart: true, + P2P_AutoBroadcast: true, + P2P_AutoSyncPeers: "Desktop,android", + P2P_AutoWatchPeers: "Desktop,android", + }; + await writeFile( + settingsPath, + JSON.stringify({ + ...createDefaultCliSettings(), + isConfigured: true, + remoteType, + liveSync: true, + periodicReplication: true, + syncOnSave: true, + ...p2pSettings, + }) + ); + const savedSettings = await readFile(settingsPath, "utf-8"); + let settingsBeforeHost: ObsidianLiveSyncSettings | undefined; + let settingsAtConnect: ObsidianLiveSyncSettings | undefined; + const connect = vi.fn(); + const runner = vi.fn(async (options, context) => { + settingsBeforeHost = cloneSettings(context.core.services.setting.currentSettings()); + connect.mockImplementation(async () => { + settingsAtConnect = cloneSettings(context.core.services.setting.currentSettings()); + throw new Error("Host connection probe complete"); + }); + return runCommand(options, { + ...context, + p2pReplicator: { + ...context.p2pReplicator!, + transportLifecycle: { isConnected: false, connect, disconnect: vi.fn() }, + }, + }); + }); + + await start("p2p-host", runner); + + expect(settingsBeforeHost).toMatchObject({ + P2P_AutoBroadcast: false, + P2P_AutoSyncPeers: "", + P2P_AutoWatchPeers: "", + }); + expect(connect).toHaveBeenCalledOnce(); + expect(settingsAtConnect).toMatchObject(p2pSettings); + expect(settingsAtConnect).toMatchObject({ liveSync: false, periodicReplication: false, syncOnSave: false }); + expect(await readFile(settingsPath, "utf-8")).toBe(savedSettings); + } + ); + + it.each(["ls", "put", "remote-set", "p2p-peers", "p2p-sync"] as const)( + "keeps automatic P2P work suspended for %s", + async (command) => { + await writeFile( + settingsPath, + JSON.stringify({ + ...createDefaultCliSettings(), + isConfigured: true, + remoteType: REMOTE_P2P, + P2P_Enabled: true, + P2P_AutoAccepting: AutoAccepting.ALL, + P2P_AutoStart: true, + P2P_AutoBroadcast: true, + P2P_AutoSyncPeers: "Desktop", + P2P_AutoWatchPeers: "Desktop", + }) + ); + const savedSettings = await readFile(settingsPath, "utf-8"); + let settingsAtCommand: ObsidianLiveSyncSettings | undefined; + const runner = vi.fn(async (_options, { core }) => { + settingsAtCommand = cloneSettings(core.services.setting.currentSettings()); + return false; + }); + + await start(command, runner); + + expect(runner).toHaveBeenCalledOnce(); + expect(settingsAtCommand).toMatchObject({ + P2P_Enabled: true, + P2P_AutoAccepting: AutoAccepting.NONE, + P2P_AutoStart: false, + P2P_AutoBroadcast: false, + P2P_AutoSyncPeers: "", + P2P_AutoWatchPeers: "", + }); + expect(await readFile(settingsPath, "utf-8")).toBe(savedSettings); + } + ); + it.each([ { command: "daemon" as const, suspendFileWatching: false }, { command: "mirror" as const, suspendFileWatching: false }, @@ -146,20 +249,25 @@ describe("CLI database preparation", () => { expect(storedPaths).toEqual([]); }); - it("unloads without starting the command when database preparation fails", async () => { - vi.spyOn(ControlService.prototype, "onReady").mockResolvedValue(false); - const unload = vi.spyOn(ControlService.prototype, "onUnload"); - const runner = vi.fn(async () => false); - const settingsBefore = await readFile(settingsPath, "utf-8"); + it.each(["daemon", "p2p-host"] as const)( + "unloads without starting %s when database preparation fails", + async (command) => { + vi.spyOn(ControlService.prototype, "onReady").mockResolvedValue(false); + const unload = vi.spyOn(ControlService.prototype, "onUnload"); + const runner = vi.fn(async () => false); + const settingsBefore = await readFile(settingsPath, "utf-8"); - await start("daemon", runner); + await start(command, runner); - expect(runner).not.toHaveBeenCalled(); - expect(unload).toHaveBeenCalledOnce(); - expect(unload.mock.invocationCallOrder[0]).toBeLessThan(vi.mocked(process.exit).mock.invocationCallOrder[0]); - expect(process.exit).toHaveBeenCalledWith(1); - expect(await readFile(settingsPath, "utf-8")).toBe(settingsBefore); - }); + expect(runner).not.toHaveBeenCalled(); + expect(unload).toHaveBeenCalledOnce(); + expect(unload.mock.invocationCallOrder[0]).toBeLessThan( + vi.mocked(process.exit).mock.invocationCallOrder[0] + ); + expect(process.exit).toHaveBeenCalledWith(1); + expect(await readFile(settingsPath, "utf-8")).toBe(settingsBefore); + } + ); it("stops the daemon when the startup scanner refuses a suspended Vault scan", async () => { await writeFile( diff --git a/src/apps/cli/main.ts b/src/apps/cli/main.ts index 3fd1ccb9..820d189a 100644 --- a/src/apps/cli/main.ts +++ b/src/apps/cli/main.ts @@ -608,8 +608,8 @@ export async function main( ? preserveStoredSetting(latestPreparedSettingsText, originalSettingsText, "useIndexedDBAdapter") : originalSettingsText; - // Capture sync settings before suspendAllSync() clobbers them. - // Used by daemon mode to restore sync behaviour after initial replication. + // Capture configured policy before temporary CLI startup suspension. + // Long-running commands restore only the settings they own. const settingsBeforeSuspend = cloneSettings(core.services.setting.currentSettings()); const durableSettingsBeforeSuspend = cloneSettings(settingsBeforeSuspend); applyStoredSetting(durableSettingsBeforeSuspend, settingsAfterLoadText, "useIndexedDBAdapter"); @@ -622,6 +622,14 @@ export async function main( syncOnFileOpen: settingsBeforeSuspend.syncOnFileOpen, syncAfterMerge: settingsBeforeSuspend.syncAfterMerge, }; + const originalP2PSettings = { + P2P_Enabled: settingsBeforeSuspend.P2P_Enabled, + P2P_AutoAccepting: settingsBeforeSuspend.P2P_AutoAccepting, + P2P_AutoStart: settingsBeforeSuspend.P2P_AutoStart, + P2P_AutoBroadcast: settingsBeforeSuspend.P2P_AutoBroadcast, + P2P_AutoSyncPeers: settingsBeforeSuspend.P2P_AutoSyncPeers, + P2P_AutoWatchPeers: settingsBeforeSuspend.P2P_AutoWatchPeers, + }; await core.services.setting.suspendAllSync(); const settingsAfterSuspend = cloneSettings(core.services.setting.currentSettings()); let readyResult = false; @@ -679,6 +687,7 @@ export async function main( p2pReplicator, settingsPath, originalSyncSettings, + originalP2PSettings, }); } finally { commandIsRunning = false; diff --git a/src/apps/cli/test-support/p2p-automatic-host.test.ts b/src/apps/cli/test-support/p2p-automatic-host.test.ts new file mode 100644 index 00000000..229fcc57 --- /dev/null +++ b/src/apps/cli/test-support/p2p-automatic-host.test.ts @@ -0,0 +1,221 @@ +import { createInterface } from "node:readline"; +import type { FilePathWithPrefix } from "@vrtmrz/livesync-commonlib/compat/common/types"; +import type { CLICommandContext, CLIOptions } from "@/apps/cli/commands/types"; +import { runCommand } from "@/apps/cli/commands/runCommand"; + +const FRAME_PREFIX = "@@LIVESYNC_P2P_AUTOMATIC@@"; +const SEED_PATH_ENV = "LIVESYNC_CLI_P2P_AUTOMATIC_SEED_PATH"; +const SEED_CONTENT_ENV = "LIVESYNC_CLI_P2P_AUTOMATIC_SEED_CONTENT"; + +type FixtureRequest = { + id: string; + action: "state" | "wait-peer" | "write" | "wait-content" | "stop"; + peerName?: string; + path?: string; + content?: string; + timeoutMs?: number; +}; + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function describeError(value: unknown): string { + return value instanceof Error ? (value.stack ?? value.message) : String(value); +} + +function emitFrame(payload: Record): void { + process.stdout.write(`${FRAME_PREFIX}${JSON.stringify(payload)}\n`); +} + +function requirePath(path: string | undefined): FilePathWithPrefix { + if (!path?.trim() || path.includes("\0")) { + throw new Error("The P2P automatic-host fixture requires a non-empty database path"); + } + return path as FilePathWithPrefix; +} + +async function storeText( + context: CLICommandContext, + path: string | undefined, + content: string | undefined +): Promise { + if (content === undefined) { + throw new Error("The P2P automatic-host fixture write request requires content"); + } + const stored = await context.core.serviceModules.databaseFileAccess.storeContent(requirePath(path), content); + if (!stored) { + throw new Error(`Could not store '${path}' in the fixture-owned database`); + } +} + +async function readText(context: CLICommandContext, path: string | undefined): Promise { + const source = await context.core.serviceModules.databaseFileAccess.fetch(requirePath(path), undefined, true); + if (!source || source.deleted) return undefined; + if (source.body.type === "text/plain") return await source.body.text(); + return new TextDecoder().decode(await source.body.arrayBuffer()); +} + +function getPolicy(context: CLICommandContext) { + const settings = context.core.services.setting.currentSettings(); + return { + P2P_Enabled: settings.P2P_Enabled, + P2P_AutoAccepting: settings.P2P_AutoAccepting, + P2P_AutoStart: settings.P2P_AutoStart, + P2P_AutoBroadcast: settings.P2P_AutoBroadcast, + P2P_AutoSyncPeers: settings.P2P_AutoSyncPeers, + P2P_AutoWatchPeers: settings.P2P_AutoWatchPeers, + }; +} + +/** Runs a persistent real P2P service with in-process controls for the E2E test. */ +export async function runP2PAutomaticHostFixture(options: CLIOptions, context: CLICommandContext): Promise { + const seedPath = process.env[SEED_PATH_ENV]; + const seedContent = process.env[SEED_CONTENT_ENV]; + if (seedPath !== undefined || seedContent !== undefined) { + await storeText(context, seedPath, seedContent); + } + + const service = context.p2pReplicator; + if (!service) { + throw new Error("The CLI did not expose its P2P service-feature result to the automatic-host fixture"); + } + const commandOutcome = runCommand(options, context).then( + (result) => ({ type: "result" as const, result }), + (error) => ({ type: "error" as const, error }) + ); + const connectedWaitStarted = Date.now(); + while (!service.transportLifecycle.isConnected) { + const earlyOutcome = await Promise.race([commandOutcome, delay(200).then(() => undefined)]); + if (earlyOutcome) { + if (earlyOutcome.type === "error") throw earlyOutcome.error; + return earlyOutcome.result; + } + if (Date.now() - connectedWaitStarted > 30000) { + throw new Error("The production p2p-host command did not connect within 30000ms"); + } + } + const input = createInterface({ + input: process.stdin, + crlfDelay: Infinity, + }); + emitFrame({ + type: "ready", + connected: service.transportLifecycle.isConnected, + policy: getPolicy(context), + }); + + const controls = (async () => { + try { + for await (const line of input) { + let request: FixtureRequest | undefined; + try { + request = JSON.parse(line) as FixtureRequest; + if (typeof request.id !== "string" || !request.id) { + throw new Error("The P2P automatic-host fixture request requires an id"); + } + + let result: unknown; + if (request.action === "state") { + result = { + connected: service.transportLifecycle.isConnected, + peers: service.peerDirectory.getPeers().map((peer) => ({ + name: peer.name, + peerId: peer.peerId, + })), + policy: getPolicy(context), + }; + } else if (request.action === "wait-peer") { + const peerName = request.peerName?.trim(); + if (!peerName) { + throw new Error("The wait-peer request requires peerName"); + } + const timeoutMs = request.timeoutMs ?? 20000; + const started = Date.now(); + let peer = service.peerDirectory.getPeers().find((candidate) => candidate.name === peerName); + while (!peer && Date.now() - started < timeoutMs) { + await delay(200); + peer = service.peerDirectory.getPeers().find((candidate) => candidate.name === peerName); + } + if (!peer) { + const known = service.peerDirectory + .getPeers() + .map((candidate) => `${candidate.name} (${candidate.peerId})`) + .join(", "); + throw new Error( + `Peer '${peerName}' was not discovered within ${timeoutMs}ms. Known peers: ${ + known || "none" + }` + ); + } + result = { name: peer.name, peerId: peer.peerId }; + } else if (request.action === "write") { + await storeText(context, request.path, request.content); + result = { path: request.path }; + } else if (request.action === "wait-content") { + const path = requirePath(request.path); + if (request.content === undefined) { + throw new Error("The wait-content request requires content"); + } + const timeoutMs = request.timeoutMs ?? 30000; + const started = Date.now(); + let actual = await readText(context, path); + while (actual !== request.content && Date.now() - started < timeoutMs) { + await delay(200); + actual = await readText(context, path); + } + if (actual !== request.content) { + throw new Error( + `Timed out waiting for '${path}' to contain the expected text within ${timeoutMs}ms; ` + + `last content: ${actual === undefined ? "missing" : JSON.stringify(actual)}` + ); + } + emitFrame({ + type: "checkpoint", + id: request.id, + action: request.action, + path, + content: actual, + }); + result = { path, content: actual }; + } else if (request.action === "stop") { + emitFrame({ + type: "response", + id: request.id, + ok: true, + result: { stopping: true }, + }); + break; + } else { + throw new Error(`Unknown P2P automatic-host fixture action: ${String(request.action)}`); + } + + emitFrame({ + type: "response", + id: request.id, + ok: true, + result, + }); + } catch (error) { + emitFrame({ + type: "response", + id: request?.id ?? "", + ok: false, + error: describeError(error), + }); + } + } + } finally { + input.close(); + } + })(); + + const outcome = await Promise.race([ + commandOutcome.then((result) => ({ type: "command" as const, result })), + controls.then(() => ({ type: "controls" as const })), + ]); + input.close(); + if (outcome.type === "controls") return true; + if (outcome.result.type === "error") throw outcome.result.error; + return outcome.result.result; +} diff --git a/src/apps/cli/test-support/p2p-lifecycle-entrypoint.ts b/src/apps/cli/test-support/p2p-lifecycle-entrypoint.ts index aac6a78e..3bd1a49b 100644 --- a/src/apps/cli/test-support/p2p-lifecycle-entrypoint.ts +++ b/src/apps/cli/test-support/p2p-lifecycle-entrypoint.ts @@ -5,6 +5,7 @@ import { createNodeStandardIo } from "@vrtmrz/livesync-commonlib/node"; import { writeStderrLine } from "@/apps/cli/cliOutput"; import { main, type CliCommandRunner } from "@/apps/cli/main"; import { parseTimeoutSeconds } from "@/apps/cli/commands/p2p"; +import { runP2PAutomaticHostFixture } from "./p2p-automatic-host.test"; import { runP2PReplicatorReplacementProbe } from "./p2p-replicator-replacement.test"; if ( @@ -16,8 +17,13 @@ if ( const standardIo = createNodeStandardIo(); const runLifecycleProbe: CliCommandRunner = async (options, context) => { + if (options.command === "p2p-host") { + return await runP2PAutomaticHostFixture(options, context); + } if (options.command !== "p2p-sync" || options.commandArgs.length < 2) { - throw new Error("The P2P lifecycle test entry requires: p2p-sync [note-path] [note-content]"); + throw new Error( + "The P2P lifecycle test entry requires: p2p-host or p2p-sync [note-path] [note-content]" + ); } const peerToken = options.commandArgs[0].trim(); if (!peerToken) { diff --git a/src/apps/cli/test-support/p2p-replicator-replacement.test.ts b/src/apps/cli/test-support/p2p-replicator-replacement.test.ts index 1f7f6447..a83a56f3 100644 --- a/src/apps/cli/test-support/p2p-replicator-replacement.test.ts +++ b/src/apps/cli/test-support/p2p-replicator-replacement.test.ts @@ -88,7 +88,7 @@ export async function runP2PReplicatorReplacementProbe( throw new Error("The CLI did not activate the initial P2P Replicator adapter"); } const compatibilityFacade = p2pReplicator.replicator; - const p2pService = await openP2PHost(core, p2pReplicator); + const p2pService = await openP2PHost(core, p2pReplicator, context.originalP2PSettings); const firstPeer = await communicateWithPeer(p2pService, targetPeer, timeoutMs); const initialised = await core.services.databaseEvents.initialiseDatabase(false, true, false); diff --git a/src/apps/cli/testdeno/deno.json b/src/apps/cli/testdeno/deno.json index 30262fa6..4d114f82 100644 --- a/src/apps/cli/testdeno/deno.json +++ b/src/apps/cli/testdeno/deno.json @@ -18,9 +18,10 @@ "test:p2p-host": "deno test --env-file=.test.env -A --no-check test-p2p-host.ts", "test:p2p-peers": "deno test --env-file=.test.env -A --no-check test-p2p-peers-local-relay.ts", "test:p2p-sync": "deno test --env-file=.test.env -A --no-check test-p2p-sync.ts", + "test:p2p-automatic-host": "deno test --env-file=.test.env -A --no-check test-p2p-automatic-host.ts", "test:p2p-replacement": "deno test --env-file=.test.env -A --no-check test-p2p-replicator-replacement.ts", "test:p2p-relay-disconnect": "deno test --env-file=.test.env -A --no-check test-p2p-relay-disconnect.ts", - "test:p2p:ci": "deno test --env-file=.test.env -A --no-check test-p2p-sync.ts test-p2p-replicator-replacement.ts test-p2p-relay-disconnect.ts", + "test:p2p:ci": "deno test --env-file=.test.env -A --no-check test-p2p-sync.ts test-p2p-automatic-host.ts test-p2p-replicator-replacement.ts test-p2p-relay-disconnect.ts", "test:p2p-three-nodes": "deno test --env-file=.test.env -A --no-check test-p2p-three-nodes-conflict.ts", "test:p2p-upload-download": "deno test --env-file=.test.env -A --no-check test-p2p-upload-download-repro.ts", "test:benchmark-contract": "deno test --env-file=.test.env -A --no-check test-benchmark-contract.ts", diff --git a/src/apps/cli/testdeno/run-cli-e2e.sh b/src/apps/cli/testdeno/run-cli-e2e.sh index af5baf0e..164a4e4a 100644 --- a/src/apps/cli/testdeno/run-cli-e2e.sh +++ b/src/apps/cli/testdeno/run-cli-e2e.sh @@ -4,12 +4,12 @@ set -eu TASK="${CLI_E2E_TASK:-test:p2p:ci}" case "$TASK" in - test:p2p-host|test:p2p-peers|test:p2p-sync|test:p2p-replacement|test:p2p-relay-disconnect|test:p2p:ci|test:p2p-three-nodes|test:p2p-upload-download) + test:p2p-host|test:p2p-peers|test:p2p-sync|test:p2p-automatic-host|test:p2p-replacement|test:p2p-relay-disconnect|test:p2p:ci|test:p2p-three-nodes|test:p2p-upload-download) exec deno task "$TASK" ;; *) echo "Unknown CLI_E2E_TASK: $TASK" >&2 - echo "Expected one of: test:p2p-host, test:p2p-peers, test:p2p-sync, test:p2p-replacement, test:p2p-relay-disconnect, test:p2p:ci, test:p2p-three-nodes, test:p2p-upload-download" >&2 + echo "Expected one of: test:p2p-host, test:p2p-peers, test:p2p-sync, test:p2p-automatic-host, test:p2p-replacement, test:p2p-relay-disconnect, test:p2p:ci, test:p2p-three-nodes, test:p2p-upload-download" >&2 exit 2 ;; esac diff --git a/src/apps/cli/testdeno/test-p2p-automatic-host.ts b/src/apps/cli/testdeno/test-p2p-automatic-host.ts new file mode 100644 index 00000000..f9b9c2a0 --- /dev/null +++ b/src/apps/cli/testdeno/test-p2p-automatic-host.ts @@ -0,0 +1,422 @@ +import { assert, assertEquals } from "@std/assert"; +import { join } from "@std/path"; +import { TempDir } from "./helpers/temp.ts"; +import { applyP2pSettings, applyP2pTestTweaks, initSettingsFile } from "./helpers/settings.ts"; +import { CLI_DIR } from "./helpers/cli.ts"; +import { maybeStartCoturn, maybeStartLocalRelay, stopCoturnIfStarted, stopLocalRelayIfStarted } from "./helpers/p2p.ts"; + +const FRAME_PREFIX = "@@LIVESYNC_P2P_AUTOMATIC@@"; +const SEED_PATH_ENV = "LIVESYNC_CLI_P2P_AUTOMATIC_SEED_PATH"; +const SEED_CONTENT_ENV = "LIVESYNC_CLI_P2P_AUTOMATIC_SEED_CONTENT"; +const HOST_BASELINE = { + path: "p2p/automatic-host-baseline.md", + content: "Saved by the persistent P2P host.", +}; +const CLIENT_A_BASELINE = { + path: "p2p/automatic-client-a-baseline.md", + content: "Saved by client A before joining.", +}; +const CLIENT_A_UPDATE = { + path: "p2p/automatic-client-a-update.md", + content: "Sent through the host auto-watch.", +}; +const HOST_UPDATE = { + path: "p2p/automatic-host-update.md", + content: "Broadcast by the host to its watcher.", +}; +const CLIENT_B_UPDATE = { + path: "p2p/automatic-client-b-update.md", + content: "Sent through the restarted host's watcher.", +}; +const CLIENT_B_DURING_OUTAGE = { + path: "p2p/automatic-client-b-during-outage.md", + content: "Retained by client B while the host restarts.", +}; + +type FixtureState = { + connected: boolean; + peers: { name: string; peerId: string }[]; + policy: Record; +}; + +type FixtureResponse = { + type: "response"; + id: string; + ok: boolean; + result?: unknown; + error?: string; +}; + +type FixtureReady = { + type: "ready"; + connected: boolean; + policy: Record; +}; + +function withTimeout(promise: Promise, timeoutMs: number, description: string): Promise { + let timer: ReturnType | undefined; + return Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out ${description} after ${timeoutMs}ms`)), timeoutMs); + }), + ]).finally(() => { + if (timer !== undefined) clearTimeout(timer); + }); +} + +class AutomaticP2PHostProcess { + #stdout = ""; + #stdoutLineBuffer = ""; + #stderr = ""; + #nextId = 0; + #pending = new Map< + string, + { + resolve: (value: FixtureResponse) => void; + reject: (reason: Error) => void; + } + >(); + #resolveReady!: (value: FixtureReady) => void; + #rejectReady!: (reason: Error) => void; + #ready = new Promise((resolve, reject) => { + this.#resolveReady = resolve; + this.#rejectReady = reject; + }); + #stdinWriter: WritableStreamDefaultWriter; + #stdoutTask: Promise; + #stderrTask: Promise; + #status: Promise; + #stopped = false; + + private constructor(readonly child: Deno.ChildProcess) { + this.#stdinWriter = child.stdin.getWriter(); + this.#stdoutTask = this.#pump(child.stdout, (text) => this.#handleStdout(text)); + this.#stderrTask = this.#pump(child.stderr, (text) => (this.#stderr += text)); + this.#status = child.status.then((status) => { + if (!this.#stopped) { + const error = new Error( + `P2P automatic-host fixture exited with code ${status.code}\n${this.diagnostics}` + ); + this.#rejectReady(error); + for (const pending of this.#pending.values()) { + pending.reject(error); + } + this.#pending.clear(); + } + return status; + }); + } + + static async start( + vaultPath: string, + settingsPath: string, + seed: { path: string; content: string } | undefined, + readyTimeoutMs: number + ): Promise { + const env: Record = {}; + if (seed) { + env[SEED_PATH_ENV] = seed.path; + env[SEED_CONTENT_ENV] = seed.content; + } + const child = new Deno.Command("node", { + args: [join(CLI_DIR, "dist", "p2p-lifecycle-test.cjs"), vaultPath, "--settings", settingsPath, "p2p-host"], + cwd: CLI_DIR, + env, + stdin: "piped", + stdout: "piped", + stderr: "piped", + }).spawn(); + const fixture = new AutomaticP2PHostProcess(child); + try { + await withTimeout(fixture.#ready, readyTimeoutMs, "waiting for the P2P automatic-host fixture to connect"); + return fixture; + } catch (error) { + await fixture.#kill().catch(() => {}); + throw new Error(`${String(error)}\n${fixture.diagnostics}`); + } + } + + get diagnostics(): string { + return `stdout:\n${this.#stdout.slice(-8000)}\nstderr:\n${this.#stderr.slice(-8000)}`; + } + + async #pump(stream: ReadableStream, consume: (text: string) => void): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (value) consume(decoder.decode(value, { stream: true })); + } + consume(decoder.decode()); + } finally { + reader.releaseLock(); + } + } + + #handleStdout(text: string): void { + this.#stdout += text; + const lines = (this.#stdoutLineBuffer + text).split(/\r?\n/); + this.#stdoutLineBuffer = lines.pop() ?? ""; + for (const line of lines) { + if (!line.startsWith(FRAME_PREFIX)) continue; + try { + const frame = JSON.parse(line.slice(FRAME_PREFIX.length)) as FixtureReady | FixtureResponse; + if (frame.type === "ready") { + this.#resolveReady(frame); + } else if (frame.type === "response") { + const pending = this.#pending.get(frame.id); + if (!pending) continue; + this.#pending.delete(frame.id); + pending.resolve(frame); + } + } catch (error) { + this.#rejectReady(new Error(`Could not parse a P2P automatic-host fixture response: ${String(error)}`)); + } + } + } + + async ready(): Promise { + return await this.#ready; + } + + async request(action: string, fields: Record = {}, timeoutMs = 30000): Promise { + await this.#ready; + const id = String(++this.#nextId); + const responsePromise = new Promise((resolve, reject) => { + this.#pending.set(id, { resolve, reject }); + }); + try { + await this.#stdinWriter.write(new TextEncoder().encode(`${JSON.stringify({ id, action, ...fields })}\n`)); + const response = await withTimeout(responsePromise, timeoutMs, `waiting for fixture action '${action}'`); + if (!response.ok) { + throw new Error(response.error ?? `Fixture action '${action}' failed`); + } + return response.result as T; + } catch (error) { + this.#pending.delete(id); + throw new Error(`${String(error)}\n${this.diagnostics}`); + } + } + + async state(): Promise { + return await this.request("state"); + } + + async waitForPeer(peerName: string, timeoutMs: number): Promise { + await this.request("wait-peer", { peerName, timeoutMs }, timeoutMs + 3000); + } + + async write(path: string, content: string): Promise { + await this.request("write", { path, content }); + } + + async waitForContent(note: { path: string; content: string }, timeoutMs: number): Promise { + await this.request( + "wait-content", + { + path: note.path, + content: note.content, + timeoutMs, + }, + timeoutMs + 3000 + ); + } + + async stop(): Promise { + if (this.#stopped) return (await this.#status).code; + try { + await this.request("stop", {}, 5000); + this.#stopped = true; + const status = await withTimeout(this.#status, 12000, "waiting for the P2P fixture to stop gracefully"); + await Promise.all([this.#stdoutTask, this.#stderrTask]); + return status.code; + } catch (error) { + await this.#kill().catch(() => {}); + throw new Error(`${String(error)}\n${this.diagnostics}`); + } + } + + async #kill(): Promise { + this.#stopped = true; + try { + this.child.kill("SIGTERM"); + } catch { + // The process may already have exited. + } + try { + await withTimeout(this.#status, 3000, "waiting for the P2P fixture to exit"); + } catch { + try { + this.child.kill("SIGKILL"); + } catch { + // The process may already have exited. + } + await this.#status; + } + await Promise.all([this.#stdoutTask, this.#stderrTask]); + } +} + +async function configureAutomaticSettings( + settingsPath: string, + peerName: string, + roomId: string, + passphrase: string, + relay: string, + turnServers: string, + autoAcceptingPeers: string, + autoSyncPeers: string, + autoWatchPeers: string +): Promise { + await initSettingsFile(settingsPath); + await applyP2pSettings( + settingsPath, + roomId, + passphrase, + "self-hosted-livesync-cli-automatic-host-tests", + relay, + "~.*", + turnServers + ); + await applyP2pTestTweaks(settingsPath, peerName, passphrase); + const settings = JSON.parse(await Deno.readTextFile(settingsPath)); + settings.P2P_AutoAccepting = 0; + settings.P2P_AutoStart = false; + settings.P2P_AutoBroadcast = true; + settings.P2P_AutoSyncPeers = autoSyncPeers; + settings.P2P_AutoWatchPeers = autoWatchPeers; + settings.P2P_AutoAcceptingPeers = autoAcceptingPeers; + settings.P2P_AutoDenyingPeers = ""; + await Deno.writeTextFile(settingsPath, JSON.stringify(settings, null, 2)); +} + +Deno.test("p2p-host: restored automatic policy syncs clients through a persistent middle peer", async () => { + const relay = Deno.env.get("RELAY") ?? "ws://localhost:4000/"; + const peersTimeoutMs = Number(Deno.env.get("PEERS_TIMEOUT") ?? "20") * 1000; + const syncTimeoutMs = Number(Deno.env.get("SYNC_TIMEOUT") ?? "60") * 1000; + const nonce = `${Date.now()}-${crypto.randomUUID().slice(0, 8)}`; + const roomId = `automatic-host-room-${nonce}`; + const passphrase = `automatic-host-pass-${nonce}`; + const hostPeerName = `p2p-auto-host-${nonce}`; + const clientAPeerName = `p2p-auto-client-a-${nonce}`; + const clientBPeerName = `p2p-auto-client-b-${nonce}`; + const useCoturn = Deno.env.get("LIVESYNC_USE_COTURN") !== "0"; + const turnServers = Deno.env.get("TURN_SERVERS") ?? (useCoturn ? "turn:127.0.0.1:3478" : "none"); + + await using workDir = await TempDir.create("livesync-cli-p2p-automatic-host"); + const hostVault = workDir.join("vault-host"); + const clientAVault = workDir.join("vault-client-a"); + const clientBVault = workDir.join("vault-client-b"); + const hostSettings = workDir.join("settings-host.json"); + const clientASettings = workDir.join("settings-client-a.json"); + const clientBSettings = workDir.join("settings-client-b.json"); + await Promise.all([ + Deno.mkdir(hostVault, { recursive: true }), + Deno.mkdir(clientAVault, { recursive: true }), + Deno.mkdir(clientBVault, { recursive: true }), + ]); + + const relayStarted = await maybeStartLocalRelay(relay); + const coturnStarted = await maybeStartCoturn(turnServers); + let host: AutomaticP2PHostProcess | undefined; + let clientA: AutomaticP2PHostProcess | undefined; + let clientB: AutomaticP2PHostProcess | undefined; + try { + const hostPeerPattern = "~p2p-auto-client-.*"; + await configureAutomaticSettings( + hostSettings, + hostPeerName, + roomId, + passphrase, + relay, + turnServers, + hostPeerPattern, + hostPeerPattern, + hostPeerPattern + ); + for (const [settingsPath, peerName] of [ + [clientASettings, clientAPeerName], + [clientBSettings, clientBPeerName], + ]) { + await configureAutomaticSettings( + settingsPath, + peerName, + roomId, + passphrase, + relay, + turnServers, + hostPeerName, + hostPeerName, + hostPeerName + ); + } + + host = await AutomaticP2PHostProcess.start(hostVault, hostSettings, HOST_BASELINE, 20000); + const initialHostState = await host.state(); + assert(initialHostState.connected, "p2p-host did not explicitly connect while P2P_AutoStart was false"); + assertEquals( + initialHostState.policy, + { + P2P_Enabled: true, + P2P_AutoAccepting: 0, + P2P_AutoStart: false, + P2P_AutoBroadcast: true, + P2P_AutoSyncPeers: hostPeerPattern, + P2P_AutoWatchPeers: hostPeerPattern, + }, + "The host did not restore its original P2P policy before serving peers" + ); + + clientA = await AutomaticP2PHostProcess.start(clientAVault, clientASettings, CLIENT_A_BASELINE, 20000); + await host.waitForPeer(clientAPeerName, peersTimeoutMs); + await clientA.waitForPeer(hostPeerName, peersTimeoutMs); + + await host.waitForContent(CLIENT_A_BASELINE, syncTimeoutMs); + await clientA.waitForContent(HOST_BASELINE, syncTimeoutMs); + + await clientA.write(CLIENT_A_UPDATE.path, CLIENT_A_UPDATE.content); + await host.waitForContent(CLIENT_A_UPDATE, syncTimeoutMs); + await host.write(HOST_UPDATE.path, HOST_UPDATE.content); + await clientA.waitForContent(HOST_UPDATE, syncTimeoutMs); + + assertEquals(await clientA.stop(), 0, "Client A did not stop cleanly before going offline"); + clientA = undefined; + + clientB = await AutomaticP2PHostProcess.start(clientBVault, clientBSettings, undefined, 20000); + await host.waitForPeer(clientBPeerName, peersTimeoutMs); + await clientB.waitForPeer(hostPeerName, peersTimeoutMs); + await clientB.waitForContent(CLIENT_A_BASELINE, syncTimeoutMs); + await clientB.waitForContent(CLIENT_A_UPDATE, syncTimeoutMs); + await clientB.waitForContent(HOST_BASELINE, syncTimeoutMs); + await clientB.waitForContent(HOST_UPDATE, syncTimeoutMs); + + await clientB.write(CLIENT_B_UPDATE.path, CLIENT_B_UPDATE.content); + await host.waitForContent(CLIENT_B_UPDATE, syncTimeoutMs); + + assertEquals(await host.stop(), 0, "The persistent host did not stop cleanly for its restart check"); + host = undefined; + await clientB.write(CLIENT_B_DURING_OUTAGE.path, CLIENT_B_DURING_OUTAGE.content); + + host = await AutomaticP2PHostProcess.start(hostVault, hostSettings, undefined, 20000); + const restartedHostState = await host.state(); + assert(restartedHostState.connected, "The restarted p2p-host did not connect"); + assertEquals( + restartedHostState.policy, + initialHostState.policy, + "The restarted host did not restore its original P2P policy" + ); + await host.waitForPeer(clientBPeerName, peersTimeoutMs); + await clientB.waitForPeer(hostPeerName, peersTimeoutMs); + await host.waitForContent(CLIENT_B_UPDATE, syncTimeoutMs); + await host.waitForContent(CLIENT_B_DURING_OUTAGE, syncTimeoutMs); + await clientB.waitForContent(HOST_BASELINE, syncTimeoutMs); + await clientB.waitForContent(HOST_UPDATE, syncTimeoutMs); + } finally { + await Promise.allSettled([host?.stop(), clientA?.stop(), clientB?.stop()].filter(Boolean) as Promise[]); + await stopLocalRelayIfStarted(relayStarted); + await stopCoturnIfStarted(coturnStarted); + } +}); diff --git a/src/apps/cli/testdeno/test_dev_deno.md b/src/apps/cli/testdeno/test_dev_deno.md index 889a4fe2..ac155785 100644 --- a/src/apps/cli/testdeno/test_dev_deno.md +++ b/src/apps/cli/testdeno/test_dev_deno.md @@ -34,6 +34,7 @@ src/apps/cli/testdeno/ test-p2p-host.ts test-p2p-peers-local-relay.ts test-p2p-sync.ts + test-p2p-automatic-host.ts test-p2p-three-nodes-conflict.ts test-p2p-upload-download-repro.ts test-e2e-two-vaults-matrix.ts @@ -72,6 +73,7 @@ Main tasks: - `deno task test:p2p-host` - `deno task test:p2p-peers` - `deno task test:p2p-sync` +- `deno task test:p2p-automatic-host` - `deno task test:p2p-three-nodes` - `deno task test:p2p-upload-download` - `deno task test:e2e-couchdb` @@ -171,6 +173,11 @@ These flows do not require Bash on the host. The S3 matrix tasks, environment va - Verifies that `p2p-sync` completes after peer discovery. +### `test-p2p-automatic-host.ts` + +- Verifies that `p2p-host` restores its configured P2P automation policy and stays connected while multiple clients synchronise through it. +- Covers automatic baseline transfer, watched updates, client disconnection, and host restart persistence without manual P2P sync commands. + ### `test-p2p-three-nodes-conflict.ts` - Uses one host and two clients. diff --git a/test/bench-network/Dockerfile.runner b/test/bench-network/Dockerfile.runner index c04677bb..180141dd 100644 --- a/test/bench-network/Dockerfile.runner +++ b/test/bench-network/Dockerfile.runner @@ -18,7 +18,7 @@ COPY src/apps/webpeer/package.json ./src/apps/webpeer/package.json RUN npm ci COPY . . -RUN LIVESYNC_CLI_TEST_SUPPORT=1 npm run build -w self-hosted-livesync-cli +RUN NODE_OPTIONS=--max-old-space-size=3072 LIVESYNC_CLI_TEST_SUPPORT=1 npm run build -w self-hosted-livesync-cli WORKDIR /workspace/src/apps/cli/testdeno @@ -30,6 +30,7 @@ RUN deno cache --lock=deno.lock \ bench-couchdb.ts \ bench-compression.ts \ test-p2p-sync.ts \ + test-p2p-automatic-host.ts \ test-p2p-replicator-replacement.ts \ test-p2p-relay-disconnect.ts diff --git a/updates.md b/updates.md index 2a25ed9f..2e8f57a9 100644 --- a/updates.md +++ b/updates.md @@ -12,6 +12,13 @@ Earlier releases remain available in the 1.0 release history, the 1.0 preview hi ## Unreleased +### CLI + +#### Fixed + +- `p2p-host` restores configured P2P automatic synchronisation, following, and change announcements after initialisation. (#1247) + - Headless peer acceptance follows the configured policy. Single-operation commands retain their automatic synchronisation suspension. + ## 1.0.36 13th October, 2026