Keep P2P consumers on the current transport

src/apps/cli/testdeno/deno.json

@@ -15,6 +15,9 @@
         "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-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-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",

src/apps/cli/testdeno/relay-disconnect-probe.mjs

@@ -0,0 +1,82 @@
+import { createRequire } from "node:module";
+import { pathToFileURL } from "node:url";
+import { RTCPeerConnection } from "werift";
+import { TrysteroReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/TrysteroReplicator";
+
+const requireFromProbe = createRequire(import.meta.url);
+const commonlibEntry = requireFromProbe.resolve("@vrtmrz/livesync-commonlib/context");
+const requireFromCommonlib = createRequire(commonlibEntry);
+const nostrEntry = requireFromCommonlib.resolve("@trystero-p2p/nostr");
+const { getRelaySockets, joinRoom, pauseRelayReconnection } = await import(pathToFileURL(nostrEntry).href);
+
+const relayUrl = process.env.RELAY ?? "ws://nostr-relay:7777/";
+const timeoutMs = Number(process.env.RELAY_TIMEOUT_MS ?? 15_000);
+
+function delay(ms) {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function waitFor(description, predicate) {
+    const deadline = Date.now() + timeoutMs;
+    while (Date.now() <= deadline) {
+        if (predicate()) return;
+        await delay(50);
+    }
+    throw new Error(`Timed out waiting for ${description}`);
+}
+
+const room = joinRoom(
+    {
+        appId: `livesync-relay-disconnect-probe-${Date.now()}`,
+        password: "local-test-only",
+        relayConfig: {
+            urls: [relayUrl],
+            manualReconnection: true,
+        },
+        rtcPolyfill: RTCPeerConnection,
+    },
+    "disconnect-probe"
+);
+
+try {
+    await waitFor("the relay WebSocket to open", () =>
+        Object.values(getRelaySockets()).some((socket) => socket.readyState === WebSocket.OPEN)
+    );
+    const originalSockets = Object.values(getRelaySockets());
+
+    const replicator = new TrysteroReplicator(
+        {},
+        {
+            close: async () => undefined,
+            dispatchConnectionStatus: async () => undefined,
+        }
+    );
+    replicator.disconnectFromServer();
+
+    await waitFor("all relay WebSockets to close", () =>
+        originalSockets.every((socket) => socket.readyState === WebSocket.CLOSED)
+    );
+    await delay(4_000);
+    if (!originalSockets.every((socket) => socket.readyState === WebSocket.CLOSED)) {
+        throw new Error("A relay WebSocket reconnected while reconnection was paused");
+    }
+
+    replicator.allowReconnection();
+    await waitFor("a replacement relay WebSocket to open", () =>
+        Object.values(getRelaySockets()).some(
+            (socket) => !originalSockets.includes(socket) && socket.readyState === WebSocket.OPEN
+        )
+    );
+
+    console.log(
+        JSON.stringify({
+            socketsClosed: originalSockets.length,
+            stayedDisconnectedWhilePaused: true,
+            reconnectedAfterResume: true,
+        })
+    );
+} finally {
+    await room.leave();
+    pauseRelayReconnection();
+    for (const socket of Object.values(getRelaySockets())) socket.close();
+}

src/apps/cli/testdeno/run-cli-e2e.sh

@@ -1,15 +1,15 @@
 #!/usr/bin/env sh
 set -eu
 
-TASK="${CLI_E2E_TASK:-test:p2p-sync}"
+TASK="${CLI_E2E_TASK:-test:p2p:ci}"
 
 case "$TASK" in
-  test:p2p-host|test:p2p-peers|test:p2p-sync|test:p2p-three-nodes|test:p2p-upload-download)
+  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)
     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-three-nodes, test:p2p-upload-download" >&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
     exit 2
     ;;
 esac

src/apps/cli/testdeno/run-compose-p2p.ts

@@ -1,7 +1,7 @@
 const repositoryRoot = await Deno.realPath(new URL("../../../../", import.meta.url));
 const composeArgs = ["compose", "-f", "test/bench-network/compose.yml"];
 const p2pEnvironment = {
-    CLI_E2E_TASK: Deno.env.get("CLI_E2E_TASK") ?? "test:p2p-sync",
+    CLI_E2E_TASK: Deno.env.get("CLI_E2E_TASK") ?? "test:p2p:ci",
     RELAY: Deno.env.get("RELAY") ?? "ws://nostr-relay:7777/",
     PEERS_TIMEOUT: Deno.env.get("PEERS_TIMEOUT") ?? "20",
     SYNC_TIMEOUT: Deno.env.get("SYNC_TIMEOUT") ?? "60",
@@ -10,6 +10,8 @@ const p2pEnvironment = {
     LIVESYNC_P2P_PEERS_RETRY: Deno.env.get("LIVESYNC_P2P_PEERS_RETRY") ?? "1",
     LIVESYNC_P2P_RELAY_READY_TIMEOUT_MS: Deno.env.get("LIVESYNC_P2P_RELAY_READY_TIMEOUT_MS") ?? "60000",
     BENCH_LIVESYNC_TEST_TEE: Deno.env.get("BENCH_LIVESYNC_TEST_TEE") ?? "0",
+    LIVESYNC_CLI_DEBUG: Deno.env.get("LIVESYNC_CLI_DEBUG") ?? "0",
+    LIVESYNC_CLI_VERBOSE: Deno.env.get("LIVESYNC_CLI_VERBOSE") ?? "0",
 };
 
 async function runDocker(args: string[], env?: Record<string, string>): Promise<Deno.CommandStatus> {

src/apps/cli/testdeno/test-p2p-relay-disconnect.ts

@@ -0,0 +1,26 @@
+import { assert, assertEquals } from "@std/assert";
+
+Deno.test("p2p lifecycle: explicit disconnect closes and pauses relay WebSockets", async () => {
+    const command = new Deno.Command("node", {
+        args: [new URL("./relay-disconnect-probe.mjs", import.meta.url).pathname],
+        env: {
+            RELAY: Deno.env.get("RELAY") ?? "ws://nostr-relay:7777/",
+            RELAY_TIMEOUT_MS: Deno.env.get("LIVESYNC_P2P_RELAY_READY_TIMEOUT_MS") ?? "15000",
+        },
+        stdout: "piped",
+        stderr: "piped",
+    });
+    const result = await command.output();
+    const stdout = new TextDecoder().decode(result.stdout).trim();
+    const stderr = new TextDecoder().decode(result.stderr).trim();
+
+    assert(result.success, `Relay disconnect probe failed\nstdout: ${stdout}\nstderr: ${stderr}`);
+    const report = JSON.parse(stdout.split("\n").at(-1) ?? "{}") as {
+        socketsClosed?: number;
+        stayedDisconnectedWhilePaused?: boolean;
+        reconnectedAfterResume?: boolean;
+    };
+    assert((report.socketsClosed ?? 0) > 0, "The probe did not observe an open relay WebSocket");
+    assertEquals(report.stayedDisconnectedWhilePaused, true);
+    assertEquals(report.reconnectedAfterResume, true);
+});

src/apps/cli/testdeno/test-p2p-replicator-replacement.ts

@@ -0,0 +1,117 @@
+import { assert, assertEquals, assertStringIncludes } from "@std/assert";
+import { join } from "@std/path";
+import { TempDir } from "./helpers/temp.ts";
+import { initSettingsFile, applyP2pSettings, applyP2pTestTweaks } from "./helpers/settings.ts";
+import { startCliInBackground } from "./helpers/backgroundCli.ts";
+import { maybeStartLocalRelay, stopLocalRelayIfStarted, maybeStartCoturn, stopCoturnIfStarted } from "./helpers/p2p.ts";
+import { CLI_DIR, runCli, sanitiseCatStdout } from "./helpers/cli.ts";
+
+const NOTE_PATH = "p2p-replicator-replacement.md";
+const NOTE_CONTENT = "Replicated after replacing the active P2P replicator.";
+
+async function runReplacementProbe(
+    vaultPath: string,
+    settingsPath: string,
+    targetPeer: string,
+    timeoutMs: number
+): Promise<{ code: number; stdout: string; stderr: string }> {
+    const command = new Deno.Command("node", {
+        args: [
+            join(CLI_DIR, "dist", "p2p-lifecycle-test.cjs"),
+            vaultPath,
+            "--settings",
+            settingsPath,
+            "p2p-sync",
+            targetPeer,
+            String(timeoutMs / 1000),
+            NOTE_PATH,
+            NOTE_CONTENT,
+        ],
+        cwd: CLI_DIR,
+        stdout: "piped",
+        stderr: "piped",
+    });
+    const result = await command.output();
+    return {
+        code: result.code,
+        stdout: new TextDecoder().decode(result.stdout),
+        stderr: new TextDecoder().decode(result.stderr),
+    };
+}
+
+Deno.test("p2p lifecycle: replacement keeps real CLI communication on the current replicator", async () => {
+    const relay = Deno.env.get("RELAY") ?? "ws://localhost:4000/";
+    const peersTimeout = Number(Deno.env.get("PEERS_TIMEOUT") ?? "20");
+    const syncTimeout = Number(Deno.env.get("SYNC_TIMEOUT") ?? "60");
+    const probeTimeoutMs = Math.max(peersTimeout, syncTimeout) * 1000;
+    const nonce = `${Date.now()}-${crypto.randomUUID().slice(0, 8)}`;
+    const roomId = Deno.env.get("ROOM_ID") ?? `replacement-room-${nonce}`;
+    const passphrase = Deno.env.get("PASSPHRASE") ?? `replacement-pass-${nonce}`;
+    const appId = "self-hosted-livesync-cli-replacement-test";
+    const hostPeerName = `p2p-replacement-host-${nonce}`;
+    const probePeerName = `p2p-replacement-probe-${nonce}`;
+    const verifierPeerName = `p2p-replacement-verifier-${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-replacement");
+    const hostVault = workDir.join("vault-host");
+    const probeVault = workDir.join("vault-probe");
+    const verifierVault = workDir.join("vault-verifier");
+    const hostSettings = workDir.join("settings-host.json");
+    const probeSettings = workDir.join("settings-probe.json");
+    const verifierSettings = workDir.join("settings-verifier.json");
+    await Promise.all([
+        Deno.mkdir(hostVault, { recursive: true }),
+        Deno.mkdir(probeVault, { recursive: true }),
+        Deno.mkdir(verifierVault, { recursive: true }),
+    ]);
+
+    const relayStarted = await maybeStartLocalRelay(relay);
+    const coturnStarted = await maybeStartCoturn(turnServers);
+    try {
+        for (const settingsPath of [hostSettings, probeSettings, verifierSettings]) {
+            await initSettingsFile(settingsPath);
+            await applyP2pSettings(settingsPath, roomId, passphrase, appId, relay, "~.*", turnServers);
+        }
+        await applyP2pTestTweaks(hostSettings, hostPeerName, passphrase);
+        await applyP2pTestTweaks(probeSettings, probePeerName, passphrase);
+        await applyP2pTestTweaks(verifierSettings, verifierPeerName, passphrase);
+
+        const host = startCliInBackground(hostVault, "--settings", hostSettings, "p2p-host");
+        try {
+            await host.waitUntilContains("P2P host is running", 20000);
+            const probe = await runReplacementProbe(probeVault, probeSettings, hostPeerName, probeTimeoutMs);
+            assert(
+                probe.code === 0,
+                `P2P replacement probe failed\nstdout: ${probe.stdout}\nstderr: ${probe.stderr}`
+            );
+            assertStringIncludes(probe.stdout, "[Probe] P2P replicator replaced");
+
+            const syncResult = await runCli(
+                verifierVault,
+                "--settings",
+                verifierSettings,
+                "p2p-sync",
+                hostPeerName,
+                String(syncTimeout)
+            );
+            assert(
+                syncResult.code === 0,
+                `Verifier P2P sync failed\nstdout: ${syncResult.stdout}\nstderr: ${syncResult.stderr}`
+            );
+
+            const catResult = await runCli(verifierVault, "--settings", verifierSettings, "cat", NOTE_PATH);
+            assert(
+                catResult.code === 0,
+                `Verifier could not read ${NOTE_PATH}\nstdout: ${catResult.stdout}\nstderr: ${catResult.stderr}`
+            );
+            assertEquals(sanitiseCatStdout(catResult.stdout).trim(), NOTE_CONTENT);
+        } finally {
+            await host.stop();
+        }
+    } finally {
+        await stopLocalRelayIfStarted(relayStarted);
+        await stopCoturnIfStarted(coturnStarted);
+    }
+});