diff --git a/docs/connected-hosts.md b/docs/connected-hosts.md index 20c86c1295..7e8b67ee62 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -74,7 +74,10 @@ registrations remain inspectable; a changed destination requires explicit identi Both daemons advertise `connectedHosts`; the shared link additionally requires `hostMultiplexing`. A completes capability and pinned-identity checks on each physical recovery before admitting channels. There is no local fallback. An older daemon continues to serve existing local -operations; requesting this feature gives update guidance. +operations; requesting this feature gives update guidance. The registry retries an older destination +with capped backoff, so it comes online once updated. Self, duplicate, changed-identity, and +authentication failures stay in `error` until an update changes the connection, `enabled`, or +`replaceIdentity`, or A restarts. Disposing a routed API releases its observations and session. It does not remove the saved host or affect sibling API scopes. Disposing the origin cascades to its derived scopes. Observations restore diff --git a/packages/client/src/node/index.ts b/packages/client/src/node/index.ts index 30d2b8744f..464d16332d 100644 --- a/packages/client/src/node/index.ts +++ b/packages/client/src/node/index.ts @@ -204,15 +204,16 @@ export function createHostTransportFactory( } export function createHostClient( connection: string, - config: Omit & HostTransportOptions, + config: Omit & HostTransportOptions, budget?: HostTransportBudget, ): DaemonClient { const target = parseHostTarget(connection); - const { webSocketAgent, ...clientConfig } = config; + const { webSocketAgent, transportFactory, ...clientConfig } = config; return new DaemonClient({ ...clientConfig, url: "ws://host-connector/ws", ...(target.password ? { password: target.password } : {}), - transportFactory: createHostTransportFactory(connection, budget, { webSocketAgent }), + transportFactory: + transportFactory ?? createHostTransportFactory(connection, budget, { webSocketAgent }), }); } diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index 4c6d79aced..264589f82e 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -784,6 +784,211 @@ test("old capabilities fail before execution and authentication errors recover t expect((await online(paseo)).hosts[0]!.serverId).toBe(b.daemon.getServerId()); }); +async function updatableDestination() { + let features: { connectedHosts: boolean; hostMultiplexing: boolean } | undefined; + let holding = false; + const held: (() => void)[] = []; + let connections = 0; + const http = createServer(); + const ws = new WebSocketServer({ server: http }); + ws.on("connection", (socket) => { + connections++; + socket.on("message", (data) => { + const frame = JSON.parse(data.toString()); + if (frame.type === "ping") socket.send(JSON.stringify({ type: "pong" })); + if (frame.type !== "hello") return; + const reply = () => + socket.send( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "srv_updated", + hostname: "updated", + version: "0.12.0", + ...(features ? { features } : {}), + }, + }, + }), + ); + if (holding) held.push(reply); + else reply(); + }); + }); + http.listen(0, "127.0.0.1"); + await once(http, "listening"); + cleanup.push(async () => { + for (const socket of ws.clients) socket.terminate(); + await new Promise((resolve) => ws.close(() => resolve())); + await new Promise((resolve) => http.close(() => resolve())); + }); + const address = http.address(); + if (!address || typeof address === "string") throw new Error("Missing destination port"); + return { + port: address.port, + connections: () => connections, + hold: () => { + holding = true; + }, + update: () => { + features = { connectedHosts: true, hostMultiplexing: true }; + holding = false; + for (const reply of held.splice(0)) reply(); + }, + }; +} +async function capabilityRetry(paseo: ReturnType): Promise { + let retryAt = ""; + await expect + .poll( + async () => { + const row = (await paseo.hosts.list()).hosts[0]; + retryAt = row?.nextRetryAt ?? ""; + return row; + }, + { timeout: 5000 }, + ) + .toMatchObject({ + status: "offline", + nextRetryAt: expect.any(String), + error: { code: "unsupported_capability" }, + }); + return Date.parse(retryAt); +} + +test("a destination updated past unsupported_capability keeps its guidance while retrying and comes online, while duplicate_host stays terminal", async () => { + const destination = await updatableDestination(); + const a = await daemon(); + const { paseo } = await api(a); + const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` }); + await capabilityRetry(paseo); + destination.hold(); + await expect + .poll(async () => (await paseo.hosts.list()).hosts[0], { timeout: 5000 }) + .toMatchObject({ status: "connecting", error: { code: "unsupported_capability" } }); + await expect(paseo.hosts.client("srv_updated")).rejects.toMatchObject({ + code: "unsupported_capability", + }); + destination.update(); + expect((await online(paseo)).hosts[0]).toMatchObject({ + registrationId: host.registrationId, + serverId: "srv_updated", + nextRetryAt: null, + error: null, + }); + await paseo.hosts.add({ connection: `ws://127.0.0.1:${destination.port}/ws` }); + await expect + .poll(async () => (await paseo.hosts.list()).hosts[1], { timeout: 5000 }) + .toMatchObject({ status: "error", nextRetryAt: null, error: { code: "duplicate_host" } }); +}); + +test("disabling or removing a host cancels its pending capability retry", async () => { + const destination = await updatableDestination(); + const a = await daemon(); + const { paseo } = await api(a); + const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` }); + let retryAt = await capabilityRetry(paseo); + let connections = destination.connections(); + await paseo.hosts.update(host.registrationId, { enabled: false }); + await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500)); + expect(destination.connections()).toBe(connections); + await paseo.hosts.update(host.registrationId, { enabled: true }); + retryAt = await capabilityRetry(paseo); + connections = destination.connections(); + await paseo.hosts.remove(host.registrationId); + await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500)); + expect(destination.connections()).toBe(connections); +}); + +import { + createDaemonChannel, + exportPublicKey, + generateKeyPair, + openDaemonSecureChannel, + type Transport, +} from "@getpaseo/relay/e2ee"; +test("a capability retry over a relay keeps the protected endpoint security it negotiated", async () => { + const keys = generateKeyPair(); + const legacyReceived: unknown[] = []; + let connections = 0; + let retryClosed = false; + const http = createServer(); + const ws = new WebSocketServer({ server: http }); + ws.on("connection", (socket) => { + connections++; + const wire: Transport = { + send: (data) => socket.send(data), + close: (code, reason) => socket.close(code, reason), + onmessage: null, + onclose: null, + onerror: null, + }; + socket.on("message", (data, isBinary) => { + const bytes = Buffer.from(data as Buffer); + wire.onmessage?.({ + data: isBinary + ? bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) + : bytes.toString(), + isBinary, + }); + }); + socket.on("close", (code, reason) => wire.onclose?.(code, reason.toString())); + if (connections > 1) { + socket.on("close", () => { + retryClosed = true; + }); + void createDaemonChannel(wire, keys, { + onmessage: (data) => legacyReceived.push(data), + }).catch(() => {}); + return; + } + const channel = openDaemonSecureChannel(wire, { + daemonKeyPair: keys, + events: { + onmessage: (data) => { + if (typeof data !== "string" || JSON.parse(data).type !== "hello") return; + void channel.then((secure) => + secure.send( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "srv_relayed", + hostname: "relayed", + version: "0.12.0", + }, + }, + }), + ), + ); + }, + }, + }); + void channel.catch(() => {}); + }); + http.listen(0, "127.0.0.1"); + await once(http, "listening"); + cleanup.push(async () => { + for (const socket of ws.clients) socket.terminate(); + await new Promise((resolve) => ws.close(() => resolve())); + await new Promise((resolve) => http.close(() => resolve())); + }); + const address = http.address(); + if (!address || typeof address === "string") throw new Error("Missing relay port"); + const a = await daemon(); + const { paseo } = await api(a); + await paseo.hosts.add({ + connection: `relay://127.0.0.1:${address.port}/srv_relayed?key=${encodeURIComponent(exportPublicKey(keys.publicKey))}`, + }); + await capabilityRetry(paseo); + await expect.poll(() => retryClosed, { timeout: 15000 }).toBe(true); + expect(legacyReceived).toEqual([]); +}); + test("an unreachable saved target becomes online after its daemon starts without being added again", async () => { const reservation = createServer(); reservation.listen(0, "127.0.0.1"); diff --git a/packages/server/src/server/connected-hosts/index.ts b/packages/server/src/server/connected-hosts/index.ts index 15b1e16b97..bfff76239c 100644 --- a/packages/server/src/server/connected-hosts/index.ts +++ b/packages/server/src/server/connected-hosts/index.ts @@ -7,10 +7,12 @@ import { z } from "zod"; import { createHostClient, createHostChannelTransport, + createHostTransportFactory, parseHostTarget, HostTransportBudget, } from "@getpaseo/client/node"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types"; import { HostSummarySchema, HostAddSchema, @@ -235,23 +237,30 @@ export class ConnectedHosts { if (route.registrationId === id) route.close(code, "Host access was disabled, replaced or removed"); } - private start(row: Saved): void { + private start( + row: Saved, + resume?: { failures: number; transports: DaemonTransportFactory }, + ): void { if (!row.enabled || this.closed) return; - let stopped = false; - const client = createHostClient( - row.connection, - { - clientId: `host-link-${this.self.serverId}-${row.registrationId}`, - connectTimeoutMs: 10000, + const transports = + resume?.transports ?? + createHostTransportFactory(row.connection, this.budget, { webSocketAgent: webSocketProxyAgent, - }, - this.budget, - ); + }); + let stopped = false; + let attempts = resume?.failures ?? 0; + let retry: ReturnType | undefined; + const client = createHostClient(row.connection, { + clientId: `host-link-${this.self.serverId}-${row.registrationId}`, + connectTimeoutMs: 10000, + transportFactory: transports, + }); const live: Live = { client, accepted: false, stop: () => { stopped = true; + clearTimeout(retry); void client.close(); }, }; @@ -285,7 +294,7 @@ export class ConnectedHosts { if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true) throw new HostError( "unsupported_capability", - "Update the destination daemon to use connected hosts.", + "Update the destination daemon to use connected hosts; retrying automatically.", ); // Pin and persist under the same writer as management to prevent discovery races. await this.mutate(async () => { @@ -304,16 +313,22 @@ export class ConnectedHosts { if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return; Object.assign(row, next); live.accepted = true; + attempts = 0; this.changed(); }); } catch (error) { if (stopped) return; - row.status = "error"; row.error = connectionFailure(error, client.authFailureReason !== null) ?? { code: "host_offline", message: "Host connection failed", }; - row.nextRetryAt = null; + if (terminalFailures.has(row.error.code)) { + row.status = "error"; + row.nextRetryAt = null; + } else { + stopped = true; + retry = this.scheduleAcceptanceRetry(row, attempts, transports); + } this.changed(); await client.close(); } @@ -329,18 +344,17 @@ export class ConnectedHosts { if (route.registrationId === row.registrationId) route.close(); row.status = state.status === "connecting" ? "connecting" : "offline"; row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null; - row.error = - state.status === "disconnected" - ? (connectionFailure( - Object.assign(new Error("Check the SSH configuration and daemon credentials"), { - code: state.failureCode, - }), - client.authFailureReason !== null, - ) ?? { - code: "host_offline", - message: "Cannot reach host. Check the connection; retrying automatically.", - }) - : null; + if (state.status === "disconnected") + row.error = connectionFailure( + Object.assign(new Error("Check the SSH configuration and daemon credentials"), { + code: state.failureCode, + }), + client.authFailureReason !== null, + ) ?? { + code: "host_offline", + message: "Cannot reach host. Check the connection; retrying automatically.", + }; + else if (row.error?.code !== "unsupported_capability") row.error = null; if (row.error?.code === "authentication_failed") { row.status = "error"; row.nextRetryAt = null; @@ -351,6 +365,18 @@ export class ConnectedHosts { }); void client.connect().catch(() => {}); } + private scheduleAcceptanceRetry( + row: Saved, + failures: number, + transports: DaemonTransportFactory, + ): ReturnType { + const delay = acceptanceRetryDelayMs(failures); + row.status = "offline"; + row.nextRetryAt = new Date(Date.now() + delay).toISOString(); + const retry = setTimeout(() => this.start(row, { failures: failures + 1, transports }), delay); + retry.unref(); + return retry; + } async openRoute(input: { serverId: string; scopeId: string; @@ -531,6 +557,20 @@ export class ConnectedHosts { } } +const ACCEPTANCE_RETRY_BASE_MS = 1500; +const ACCEPTANCE_RETRY_MAX_MS = 30_000; + +function acceptanceRetryDelayMs(failures: number): number { + return Math.min(ACCEPTANCE_RETRY_BASE_MS * 2 ** failures, ACCEPTANCE_RETRY_MAX_MS); +} + +const terminalFailures = new Set([ + "server_id_mismatch", + "self_host", + "duplicate_host", + "authentication_failed", +]); + function connectionFailure(error: unknown, authenticationFailed: boolean): HostSummary["error"] { if (error instanceof HostError) return { code: error.code, message: error.message }; if (error instanceof Error && "code" in error && error.code === "authentication_failed")