diff --git a/packages/server/src/server/relay-transport.latency.e2e.test.ts b/packages/server/src/server/relay-transport.latency.e2e.test.ts new file mode 100644 index 0000000000..84404da0c1 --- /dev/null +++ b/packages/server/src/server/relay-transport.latency.e2e.test.ts @@ -0,0 +1,61 @@ +import { execFile } from "node:child_process"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { expect, test } from "vitest"; + +const run = promisify(execFile); + +// Opt in on Linux with passwordless sudo, unshare, ip, mount, and tc available: +// RUN_RELAY_LATENCY_E2E=1 npx vitest run src/server/relay-transport.latency.e2e.test.ts +// All networking and the hosts-file mount are private to the child namespaces. +test.runIf(process.platform === "linux" && process.env.RUN_RELAY_LATENCY_E2E === "1")( + "connects relay control and data over a slow dual-stack route with an unreachable address", + async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "paseo-relay-latency-")); + try { + const hosts = path.join(directory, "hosts"); + await writeFile( + hosts, + `${await readFile("/etc/hosts", "utf8")}\n127.0.0.1 relay-latency.test\n2001:db8:1::1 relay-latency.test\n`, + ); + const fixture = fileURLToPath( + new URL("./test-utils/relay-latency-fixture.ts", import.meta.url), + ); + for (const order of ["ipv4first", "ipv6first"]) { + const { stdout } = await run( + "sudo", + [ + "-n", + "unshare", + "--mount", + "--net", + "sh", + "-ec", + `mount --make-rprivate / +mount --bind "$1" /etc/hosts +ip link set lo up +ip link add relay-test type dummy +ip addr add 192.0.2.2/32 dev relay-test +ip -6 addr add 2001:db8::2/128 dev relay-test +ip link set relay-test up +tc qdisc add dev lo root netem delay 350ms +exec "$2" --import tsx "$3" "$4"`, + "sh", + hosts, + process.execPath, + fixture, + order, + ], + { timeout: 20_000 }, + ); + expect(stdout).toContain("relay data round trip passed"); + } + } finally { + await rm(directory, { recursive: true, force: true }); + } + }, + 45_000, +); diff --git a/packages/server/src/server/relay-transport.ts b/packages/server/src/server/relay-transport.ts index f7b73a5367..113cce605c 100644 --- a/packages/server/src/server/relay-transport.ts +++ b/packages/server/src/server/relay-transport.ts @@ -56,7 +56,13 @@ type ControlMessage = const CONTROL_PING_INTERVAL_MS = 10_000; const CONTROL_STALE_TIMEOUT_MS = 30_000; const CONTROL_READY_TIMEOUT_MS = 8_000; -const RELAY_WEBSOCKET_OPTIONS = { handshakeTimeout: 10_000, perMessageDeflate: false } as const; +const RELAY_WEBSOCKET_OPTIONS = { + handshakeTimeout: 10_000, + perMessageDeflate: false, + // Node cancels each non-final address when this expires, even if the next + // address is unreachable. Allow high-latency routes time to connect. + autoSelectFamilyAttemptTimeout: 5_000, +} as const; function createDefaultRelayWebSocket(url: string): RelayWebSocketLike { return new WebSocket(url, RELAY_WEBSOCKET_OPTIONS); diff --git a/packages/server/src/server/test-utils/relay-latency-fixture.ts b/packages/server/src/server/test-utils/relay-latency-fixture.ts new file mode 100644 index 0000000000..5abcccf389 --- /dev/null +++ b/packages/server/src/server/test-utils/relay-latency-fixture.ts @@ -0,0 +1,75 @@ +import assert from "node:assert/strict"; +import dns from "node:dns"; +import { once } from "node:events"; +import net from "node:net"; +import { Writable } from "node:stream"; +import pino from "pino"; +import { WebSocketServer } from "ws"; +import { startRelayTransport } from "../relay-transport.js"; + +// Match the reported runtime even on Node versions with a newer default. +net.setDefaultAutoSelectFamilyAttemptTimeout(250); +const order = process.argv[2]; +assert.ok(order === "ipv4first" || order === "ipv6first"); +dns.setDefaultResultOrder(order); + +const logs: string[] = []; +const logger = pino( + { level: "debug" }, + new Writable({ + write(chunk, _encoding, callback) { + logs.push(chunk.toString()); + callback(); + }, + }), +); +const relay = new WebSocketServer({ host: "127.0.0.1", port: 0 }); +await once(relay, "listening"); +const address = relay.address(); +assert.ok(address && typeof address !== "string"); +relay.on("connection", (socket, request) => { + const url = new URL(request.url!, "http://relay-latency.test"); + if (url.searchParams.has("connectionId")) { + socket.on("message", (data) => socket.send(data)); + } else { + socket.send(JSON.stringify({ type: "sync", connectionIds: ["latency-client"] })); + } +}); + +let complete!: () => void; +let fail!: (error: Error) => void; +const delivered = new Promise((resolve, reject) => { + complete = resolve; + fail = reject; +}); +const deadline = setTimeout(() => fail(new Error(logs.join(""))), 8_000); +const transport = startRelayTransport({ + logger, + relayEndpoint: `relay-latency.test:${address.port}`, + relayUseTls: false, + serverId: "latency-server", + attachSocket: async (socket) => { + socket.on("message", (data) => { + try { + assert.equal(String(data), "latency round trip"); + complete(); + } catch (error) { + fail(error as Error); + } + }); + socket.send("latency round trip"); + }, +}); + +try { + await delivered; + assert.ok(logs.some((line) => JSON.parse(line).msg === "relay_control_connected")); + assert.ok(logs.some((line) => JSON.parse(line).msg === "relay_data_connected")); + assert.ok(!logs.some((line) => JSON.parse(line).msg === "relay_error")); + console.log("relay data round trip passed"); +} finally { + clearTimeout(deadline); + await transport.stop(); + for (const socket of relay.clients) socket.terminate(); + await new Promise((resolve) => relay.close(() => resolve())); +}