diff --git a/src/apps/cli/commands/p2p.ts b/src/apps/cli/commands/p2p.ts index 2ba76dd..e77a0cd 100644 --- a/src/apps/cli/commands/p2p.ts +++ b/src/apps/cli/commands/p2p.ts @@ -4,12 +4,21 @@ import type { ServiceContext } from "@lib/services/base/ServiceBase"; import { LiveSyncTrysteroReplicator } from "@lib/replication/trystero/LiveSyncTrysteroReplicator"; import { compatGlobal } from "@lib/common/coreEnvFunctions.ts"; import { LiveSyncError } from "@lib/common/LSError"; +import { getPeerConnectionStats } from "@lib/rpc/transports/DiagRTCPeerConnections.utils"; +import { appendFile } from "node:fs/promises"; type CLIP2PPeer = { peerId: string; name: string; }; +type CandidateSummary = { + id: string | "unknown"; + candidateType: string | "unknown"; + protocol: string | "unknown"; + relayProtocol: string | "unknown"; +}; + function delay(ms: number): Promise { return new Promise((resolve) => compatGlobal.setTimeout(resolve, ms)); } @@ -81,6 +90,74 @@ function resolvePeer(peers: CLIP2PPeer[], peerToken: string): CLIP2PPeer | undef return undefined; } +function getReportValue( + report: Record | undefined, + key: string +): T | "unknown" { + const value = report?.[key]; + return typeof value === "string" || typeof value === "number" ? (value as T) : "unknown"; +} + +function summariseCandidate(reports: unknown[], candidateId: string | "unknown"): CandidateSummary | undefined { + if (candidateId === "unknown") { + return undefined; + } + const report = reports.map((r) => r as Record).find((r) => r.id === candidateId); + if (!report) { + return undefined; + } + return { + id: candidateId, + candidateType: getReportValue(report, "candidateType"), + protocol: getReportValue(report, "protocol"), + relayProtocol: getReportValue(report, "relayProtocol"), + }; +} + +async function writePeerConnectionStatsIfRequested( + replicator: LiveSyncTrysteroReplicator, + peer: CLIP2PPeer +): Promise { + const outputPath = process.env.LIVESYNC_P2P_STATS_JSONL?.trim(); + if (!outputPath) { + return; + } + + const peerConnection = replicator.rawHost?.room?.getPeers()[peer.peerId]; + const stats = peerConnection ? await getPeerConnectionStats(`cli-p2p-${peer.peerId}`, peerConnection) : undefined; + const localCandidate = summariseCandidate(stats?.reports ?? [], stats?.localCandidateId ?? "unknown"); + const remoteCandidate = summariseCandidate(stats?.reports ?? [], stats?.remoteCandidateId ?? "unknown"); + const selectedPath = + localCandidate && remoteCandidate + ? `${localCandidate.candidateType}<->${remoteCandidate.candidateType}` + : "unknown"; + + const payload = { + generatedAt: new Date().toISOString(), + command: "p2p-sync", + peerId: peer.peerId, + peerName: peer.name, + candidatePathCollected: !!stats?.selectedPair, + selectedPath, + selectedPair: stats + ? { + id: stats.selectedPairId, + state: stats.state, + currentRoundTripTime: stats.currentRoundTripTime, + totalRoundTripTime: stats.totalRoundTripTime, + requestsSent: stats.requestsSent, + responsesReceived: stats.responsesReceived, + packetsDiscardedOnSend: stats.packetsDiscardedOnSend, + bytesSent: stats.bytesSent, + bytesReceived: stats.bytesReceived, + } + : undefined, + localCandidate, + remoteCandidate, + }; + await appendFile(outputPath, `${JSON.stringify(payload)}\n`, "utf8"); +} + export async function syncWithPeer( core: LiveSyncBaseCore, peerToken: string, @@ -118,6 +195,7 @@ export async function syncWithPeer( : LiveSyncError.fromError(err ?? "P2P sync failed while requesting remote sync"); } + await writePeerConnectionStatsIfRequested(replicator, targetPeer); return targetPeer; } finally { await replicator.close(); diff --git a/src/apps/cli/testdeno/bench-network-cases.ts b/src/apps/cli/testdeno/bench-network-cases.ts index c4d5072..c460bd1 100644 --- a/src/apps/cli/testdeno/bench-network-cases.ts +++ b/src/apps/cli/testdeno/bench-network-cases.ts @@ -65,6 +65,10 @@ function buildCases(): BenchmarkCase[] { ...base, BENCH_CASE: "p2p-direct-local", BENCH_TURN_SERVERS: "", + BENCH_SIMULATION_TIER: "1", + BENCH_NETWORK_PROFILE: "local-direct", + BENCH_NETWORK_MODEL: "local-runner-webrtc", + BENCH_P2P_CANDIDATE_PATH_VERIFICATION: "turn-disabled-but-selected-ice-pair-not-collected", }, }, { @@ -125,6 +129,11 @@ function buildCases(): BenchmarkCase[] { ...base, BENCH_CASE: "p2p-smartphone-vpn-direct", BENCH_TURN_SERVERS: "", + BENCH_SIMULATION_TIER: "unmeasured", + BENCH_NETWORK_PROFILE: "smartphone-vpn-direct-placeholder", + BENCH_NETWORK_MODEL: "local-runner-no-netem", + BENCH_P2P_CANDIDATE_PATH_VERIFICATION: + "structural-placeholder-only; selected ICE pair may be collected, but the path is not shaped", }, }, { @@ -137,6 +146,11 @@ function buildCases(): BenchmarkCase[] { ...base, BENCH_CASE: "p2p-user-turn", BENCH_TURN_SERVERS: localTurnServers, + BENCH_SIMULATION_TIER: "1", + BENCH_NETWORK_PROFILE: "local-turn-fallback", + BENCH_NETWORK_MODEL: "local-runner-webrtc-turn-configured", + BENCH_P2P_CANDIDATE_PATH_VERIFICATION: + "turn-configured; selected ICE pair may still be direct or relayed, so interpret the recorded candidate types", }, }, ]; diff --git a/src/apps/cli/testdeno/bench-p2p.ts b/src/apps/cli/testdeno/bench-p2p.ts index 2dfb2f5..2bbf4e5 100644 --- a/src/apps/cli/testdeno/bench-p2p.ts +++ b/src/apps/cli/testdeno/bench-p2p.ts @@ -27,6 +27,42 @@ type BenchmarkConfig = { binSizeBytes: number; peersTimeoutSeconds: number; syncTimeoutSeconds: number; + simulationTier: string; + networkProfile: string; + networkModel: string; + candidatePathVerification: string; +}; + +type P2PConnectionStats = { + generatedAt: string; + command: string; + peerId: string; + peerName: string; + candidatePathCollected: boolean; + selectedPath: string; + selectedPair?: { + id: string; + state: string; + currentRoundTripTime: number | "unknown"; + totalRoundTripTime: number | "unknown"; + requestsSent: number | "unknown"; + responsesReceived: number | "unknown"; + packetsDiscardedOnSend: number | "unknown"; + bytesSent: number | "unknown"; + bytesReceived: number | "unknown"; + }; + localCandidate?: { + id: string; + candidateType: string; + protocol: string; + relayProtocol: string; + }; + remoteCandidate?: { + id: string; + candidateType: string; + protocol: string; + relayProtocol: string; + }; }; function readEnvString(name: string, fallback: string): string { @@ -84,6 +120,10 @@ function buildConfig(): BenchmarkConfig { binSizeBytes: Math.floor(readEnvNumber("BENCH_BIN_SIZE_BYTES", 100 * 1024)), peersTimeoutSeconds: readEnvNumber("BENCH_PEERS_TIMEOUT", 20), syncTimeoutSeconds: readEnvNumber("BENCH_SYNC_TIMEOUT", 240), + simulationTier: readEnvString("BENCH_SIMULATION_TIER", "1"), + networkProfile: readEnvString("BENCH_NETWORK_PROFILE", "local-direct"), + networkModel: readEnvString("BENCH_NETWORK_MODEL", "local-runner-webrtc"), + candidatePathVerification: readEnvString("BENCH_P2P_CANDIDATE_PATH_VERIFICATION", "not-collected"), }; } @@ -112,6 +152,22 @@ function pickSampleFiles(entries: DatasetEntry[]): DatasetEntry[] { return [...unique.values()]; } +async function readLatestP2PConnectionStats(statsPath: string): Promise { + try { + const text = await Deno.readTextFile(statsPath); + const lines = text + .split(/\r?\n/) + .map((line) => line.trim()) + .filter((line) => line.length > 0); + if (lines.length === 0) { + return undefined; + } + return JSON.parse(lines[lines.length - 1]) as P2PConnectionStats; + } catch { + return undefined; + } +} + async function main(): Promise { const config = buildConfig(); const resultPath = readOptionalResultPath(); @@ -124,126 +180,153 @@ async function main(): Promise { const clientVault = workDir.join("vault-client"); const hostSettings = workDir.join("settings-host.json"); const clientSettings = workDir.join("settings-client.json"); + const p2pStatsPath = workDir.join("p2p-connection-stats.jsonl"); + const previousStatsPath = Deno.env.get("LIVESYNC_P2P_STATS_JSONL"); + Deno.env.set("LIVESYNC_P2P_STATS_JSONL", p2pStatsPath); - await Promise.all([ - Deno.mkdir(hostVault, { recursive: true }), - Deno.mkdir(clientVault, { recursive: true }), - initSettingsFile(hostSettings), - initSettingsFile(clientSettings), - ]); - - await Promise.all([ - applyP2pSettings( - hostSettings, - config.roomId, - config.passphrase, - config.appId, - config.relay, - "~.*", - config.turnServers - ), - applyP2pSettings( - clientSettings, - config.roomId, - config.passphrase, - config.appId, - config.relay, - "~.*", - config.turnServers - ), - ]); - - await Promise.all([ - applyP2pTestTweaks(hostSettings, "p2p-bench-host", config.passphrase), - applyP2pTestTweaks(clientSettings, "p2p-bench-client", config.passphrase), - ]); - - const seedFiles = await createDeterministicDataset({ - rootDir: hostVault, - datasetDirName: config.datasetDirName, - seed: config.datasetSeed, - mdCount: config.mdFileCount, - mdMinSizeBytes: config.mdMinSizeBytes, - mdMaxSizeBytes: config.mdMaxSizeBytes, - binCount: config.binFileCount, - binSizeBytes: config.binSizeBytes, - }); - - const mirrorStart = nowMs(); - await runCliOrFail(hostVault, "--settings", hostSettings, "mirror"); - const mirrorElapsed = nowMs() - mirrorStart; - - const host = startCliInBackground(hostVault, "--settings", hostSettings, "p2p-host"); try { - const hostReadyStart = nowMs(); - await host.waitUntilContains("P2P host is running", 20000); - const hostReadyElapsed = nowMs() - hostReadyStart; + await Promise.all([ + Deno.mkdir(hostVault, { recursive: true }), + Deno.mkdir(clientVault, { recursive: true }), + initSettingsFile(hostSettings), + initSettingsFile(clientSettings), + ]); - const peerDiscoveryCommandStart = nowMs(); - const peer = await discoverPeer(clientVault, clientSettings, config.peersTimeoutSeconds); - const peerDiscoveryCommandElapsed = nowMs() - peerDiscoveryCommandStart; + await Promise.all([ + applyP2pSettings( + hostSettings, + config.roomId, + config.passphrase, + config.appId, + config.relay, + "~.*", + config.turnServers + ), + applyP2pSettings( + clientSettings, + config.roomId, + config.passphrase, + config.appId, + config.relay, + "~.*", + config.turnServers + ), + ]); - const syncStart = nowMs(); - await runCliOrFail( - clientVault, - "--settings", - clientSettings, - "p2p-sync", - peer.id, - String(config.syncTimeoutSeconds) - ); - const syncElapsed = nowMs() - syncStart; + await Promise.all([ + applyP2pTestTweaks(hostSettings, "p2p-bench-host", config.passphrase), + applyP2pTestTweaks(clientSettings, "p2p-bench-client", config.passphrase), + ]); - const sampleFiles = pickSampleFiles(seedFiles.entries); - for (const sample of sampleFiles) { - const pulledPath = workDir.join(`pulled-${sample.relativePath.replaceAll("/", "_")}`); - await runCliOrFail(clientVault, "--settings", clientSettings, "pull", sample.relativePath, pulledPath); - await assertFilesEqual( - sample.absolutePath, - pulledPath, - `sample file mismatch after sync: ${sample.relativePath}` - ); - } - - const result = { - caseName: config.caseName, - mode: "p2p-cli-benchmark", - relay: config.relay, - turnServers: config.turnServers, - turnEnabled: config.turnServers.trim().length > 0, - appId: config.appId, - roomId: config.roomId, - datasetSeed: config.datasetSeed, + const seedFiles = await createDeterministicDataset({ + rootDir: hostVault, datasetDirName: config.datasetDirName, - peerId: peer.id, - peerName: peer.name, - totalFiles: seedFiles.totalFiles, - totalBytes: seedFiles.totalBytes, - mdFileCount: seedFiles.mdCount, - binFileCount: seedFiles.binCount, - mirrorElapsedMs: Number(mirrorElapsed.toFixed(1)), - hostReadyElapsedMs: Number(hostReadyElapsed.toFixed(1)), - peerDiscoveryTimeoutSeconds: config.peersTimeoutSeconds, - peerDiscoveryCommandElapsedMs: Number(peerDiscoveryCommandElapsed.toFixed(1)), - peerDiscoveryNote: - "p2p-peers waits for the requested timeout before printing discovered peers, so this is command duration, not first-peer latency.", - syncElapsedMs: Number(syncElapsed.toFixed(1)), - throughputBytesPerSec: Number((seedFiles.totalBytes / (syncElapsed / 1000)).toFixed(2)), - throughputMiBPerSec: Number((seedFiles.totalBytes / (syncElapsed / 1000) / 1024 / 1024).toFixed(4)), - }; + seed: config.datasetSeed, + mdCount: config.mdFileCount, + mdMinSizeBytes: config.mdMinSizeBytes, + mdMaxSizeBytes: config.mdMaxSizeBytes, + binCount: config.binFileCount, + binSizeBytes: config.binSizeBytes, + }); - if (resultPath) { - await Deno.writeTextFile(resultPath, JSON.stringify(result, null, 2)); + const mirrorStart = nowMs(); + await runCliOrFail(hostVault, "--settings", hostSettings, "mirror"); + const mirrorElapsed = nowMs() - mirrorStart; + + const host = startCliInBackground(hostVault, "--settings", hostSettings, "p2p-host"); + try { + const hostReadyStart = nowMs(); + await host.waitUntilContains("P2P host is running", 20000); + const hostReadyElapsed = nowMs() - hostReadyStart; + + const peerDiscoveryCommandStart = nowMs(); + const peer = await discoverPeer(clientVault, clientSettings, config.peersTimeoutSeconds); + const peerDiscoveryCommandElapsed = nowMs() - peerDiscoveryCommandStart; + + const syncStart = nowMs(); + await runCliOrFail( + clientVault, + "--settings", + clientSettings, + "p2p-sync", + peer.id, + String(config.syncTimeoutSeconds) + ); + const syncElapsed = nowMs() - syncStart; + + const sampleFiles = pickSampleFiles(seedFiles.entries); + for (const sample of sampleFiles) { + const pulledPath = workDir.join(`pulled-${sample.relativePath.replaceAll("/", "_")}`); + await runCliOrFail(clientVault, "--settings", clientSettings, "pull", sample.relativePath, pulledPath); + await assertFilesEqual( + sample.absolutePath, + pulledPath, + `sample file mismatch after sync: ${sample.relativePath}` + ); + } + + const p2pConnectionStats = await readLatestP2PConnectionStats(p2pStatsPath); + const result = { + caseName: config.caseName, + mode: "p2p-cli-benchmark", + relay: config.relay, + turnServers: config.turnServers, + turnEnabled: config.turnServers.trim().length > 0, + simulationTier: config.simulationTier, + networkProfile: config.networkProfile, + networkModel: config.networkModel, + p2pCandidatePathVerified: p2pConnectionStats?.candidatePathCollected === true, + p2pCandidatePathVerification: p2pConnectionStats?.candidatePathCollected + ? "selected ICE candidate pair collected from RTCPeerConnection.getStats" + : config.candidatePathVerification, + p2pCandidatePathNote: p2pConnectionStats?.candidatePathCollected + ? "The selected ICE candidate pair was collected by the CLI benchmark. Interpret the path from the candidate types; do not infer TURN use from configuration alone." + : config.turnServers.trim().length > 0 + ? "TURN is configured, so the selected WebRTC path may be direct, server-reflexive, or relayed. The selected ICE candidate pair was not exported by this run." + : "TURN is disabled, so a TURN-relayed path is not expected. The selected ICE candidate pair was not exported by this run.", + p2pConnectionStats, + appId: config.appId, + roomId: config.roomId, + datasetSeed: config.datasetSeed, + datasetDirName: config.datasetDirName, + peerId: peer.id, + peerName: peer.name, + totalFiles: seedFiles.totalFiles, + totalBytes: seedFiles.totalBytes, + mdFileCount: seedFiles.mdCount, + binFileCount: seedFiles.binCount, + mirrorElapsedMs: Number(mirrorElapsed.toFixed(1)), + hostReadyElapsedMs: Number(hostReadyElapsed.toFixed(1)), + peerDiscoveryTimeoutSeconds: config.peersTimeoutSeconds, + peerDiscoveryCommandElapsedMs: Number(peerDiscoveryCommandElapsed.toFixed(1)), + peerDiscoveryNote: + "p2p-peers waits for the requested timeout before printing discovered peers, so this is command duration, not first-peer latency.", + syncElapsedMs: Number(syncElapsed.toFixed(1)), + throughputBytesPerSec: Number((seedFiles.totalBytes / (syncElapsed / 1000)).toFixed(2)), + throughputMiBPerSec: Number((seedFiles.totalBytes / (syncElapsed / 1000) / 1024 / 1024).toFixed(4)), + }; + + if (resultPath) { + await Deno.writeTextFile(resultPath, JSON.stringify(result, null, 2)); + } + + console.log(JSON.stringify(result, null, 2)); + console.error( + `[Benchmark] mirrored ${seedFiles.totalFiles} files (${formatBytes( + seedFiles.totalBytes + )}) in ${formatMs(mirrorElapsed)}, ` + + `synced in ${formatMs(syncElapsed)} ` + + `(${result.throughputBytesPerSec} B/s, ${result.throughputMiBPerSec} MiB/s)` + ); + } finally { + await host.stop(); } - - console.log(JSON.stringify(result, null, 2)); - console.error( - `[Benchmark] mirrored ${seedFiles.totalFiles} files (${formatBytes(seedFiles.totalBytes)}) in ${formatMs(mirrorElapsed)}, ` + - `synced in ${formatMs(syncElapsed)} ` + - `(${result.throughputBytesPerSec} B/s, ${result.throughputMiBPerSec} MiB/s)` - ); } finally { - await host.stop(); + if (previousStatsPath === undefined) { + Deno.env.delete("LIVESYNC_P2P_STATS_JSONL"); + } else { + Deno.env.set("LIVESYNC_P2P_STATS_JSONL", previousStatsPath); + } await stopCoturnIfStarted(coturnStarted); await stopLocalRelayIfStarted(relayStarted); } diff --git a/test/bench-network/README.md b/test/bench-network/README.md index afc6f51..a2545d7 100644 --- a/test/bench-network/README.md +++ b/test/bench-network/README.md @@ -65,6 +65,11 @@ path: Use the CouchDB result as the remote-store baseline and the P2P result as the direct-transfer comparison. The Nostr relay is used for signalling in the P2P case, but synchronised note content is transferred over the WebRTC DataChannel. +The P2P result JSON records the selected WebRTC ICE candidate pair when the CLI +can collect it from `RTCPeerConnection.getStats()`. Interpret P2P paths from +the recorded candidate types rather than from TURN configuration alone. Do not +report P2P runs as Tier 2 constrained-network measurements until host and +client are captured under an equivalent shaped topology. ## Dataset and latency controls @@ -160,3 +165,5 @@ docker compose -f test/bench-network/compose.yml --profile shim run --rm bench-r The benchmark result records `simulationTier`, `networkProfile`, and `networkModel`. The shim also writes its applied `tc qdisc`, route, and interface state under `test/bench-network/bench-results/`. +This shim currently measures the CouchDB path only. It does not shape or verify +the WebRTC P2P data path.