refactor: extract host runtime controller for connection management

This commit is contained in:
Mohamed Boudra
2026-02-19 21:19:51 +07:00
parent bff1a5accc
commit f8e3204572
9 changed files with 1077 additions and 215 deletions
@@ -1,155 +1,38 @@
import { useEffect } from "react";
import { AppState } from "react-native";
import { SessionProvider } from "@/contexts/session-context";
import { useDaemonRegistry, type HostProfile } from "@/contexts/daemon-registry-context";
import { useDaemonConnections } from "@/contexts/daemon-connections-context";
import {
useDaemonRegistry,
type HostConnection,
type HostProfile,
} from "@/contexts/daemon-registry-context";
import {
buildDaemonWebSocketUrl,
buildRelayWebSocketUrl,
} from "@/utils/daemon-endpoints";
import { measureConnectionLatency } from "@/utils/test-daemon-connection";
import {
selectBestConnection,
type ConnectionCandidate,
type ConnectionProbeState,
} from "@/utils/connection-selection";
import { useQueries } from "@tanstack/react-query";
import { useMemo } from "react";
import type { ActiveConnection } from "@/contexts/daemon-connections-context";
type Candidate = {
connection: HostConnection;
connectionId: string;
url: string;
activeConnection: ActiveConnection;
daemonPublicKeyB64?: string;
};
function sortConnectionsByPreference<T extends { id: string }>(
connections: T[],
preferredId: string | null
): T[] {
if (!preferredId) return connections;
const idx = connections.findIndex((c) => c.id === preferredId);
if (idx === -1) return connections;
return [connections[idx]!, ...connections.slice(0, idx), ...connections.slice(idx + 1)];
}
function buildCandidates(host: HostProfile): Candidate[] {
const preferred = host.preferredConnectionId ?? null;
const connections = sortConnectionsByPreference(host.connections, preferred);
const out: Candidate[] = [];
for (const conn of connections) {
if (conn.type === "direct") {
out.push({
connection: conn,
connectionId: conn.id,
url: buildDaemonWebSocketUrl(conn.endpoint),
activeConnection: {
type: "direct",
endpoint: conn.endpoint,
display: conn.endpoint,
},
});
continue;
}
out.push({
connection: conn,
connectionId: conn.id,
url: buildRelayWebSocketUrl({
endpoint: conn.relayEndpoint,
serverId: host.serverId,
}),
activeConnection: { type: "relay", endpoint: conn.relayEndpoint, display: "relay" },
daemonPublicKeyB64: conn.daemonPublicKeyB64,
});
}
return out;
}
getHostRuntimeStore,
useHostRuntimeSession,
} from "@/runtime/host-runtime";
import { toDaemonConnectionUpdateFromRuntime } from "@/runtime/host-runtime-bridge";
function ManagedDaemonSession({ daemon }: { daemon: HostProfile }) {
const candidates = useMemo(() => buildCandidates(daemon), [daemon]);
const latencyQueries = useQueries({
queries: candidates.map((candidate) => ({
queryKey: ["connection-selection-latency", daemon.serverId, candidate.connectionId],
queryFn: () =>
measureConnectionLatency(candidate.connection, {
serverId: daemon.serverId,
}),
refetchInterval: 10_000,
staleTime: 9_000,
gcTime: 60_000,
retry: 1,
})),
});
const { snapshot, client } = useHostRuntimeSession(daemon.serverId);
const { connectionStates, updateConnectionStatus } = useDaemonConnections();
const hasConnectionRecord = connectionStates.has(daemon.serverId);
const probeByConnectionId = useMemo(() => {
const next = new Map<string, ConnectionProbeState>();
useEffect(() => {
if (!hasConnectionRecord) {
return;
}
updateConnectionStatus(
daemon.serverId,
toDaemonConnectionUpdateFromRuntime(snapshot)
);
}, [daemon.serverId, hasConnectionRecord, snapshot, updateConnectionStatus]);
candidates.forEach((candidate, index) => {
const query = latencyQueries[index];
if (!query) {
next.set(candidate.connectionId, { status: "pending", latencyMs: null });
return;
}
if (query.isSuccess && typeof query.data === "number") {
next.set(candidate.connectionId, {
status: "available",
latencyMs: query.data,
});
return;
}
if (query.isError) {
next.set(candidate.connectionId, { status: "unavailable", latencyMs: null });
return;
}
next.set(candidate.connectionId, { status: "pending", latencyMs: null });
});
return next;
}, [candidates, latencyQueries]);
const candidateInputs = useMemo<ConnectionCandidate[]>(
() =>
candidates.map((candidate) => ({
connectionId: candidate.connectionId,
connection: candidate.connection,
})),
[candidates]
);
const activeConnectionId = useMemo(
() =>
selectBestConnection({
candidates: candidateInputs,
preferredConnectionId: daemon.preferredConnectionId,
probeByConnectionId,
}),
[candidateInputs, daemon.preferredConnectionId, probeByConnectionId]
);
const active =
candidates.find((candidate) => candidate.connectionId === activeConnectionId) ?? null;
const activeUrl = active?.url ?? null;
if (!activeUrl) {
if (!snapshot?.activeConnection || !client) {
return null;
}
return (
<SessionProvider
key={daemon.serverId}
serverUrl={activeUrl}
serverId={daemon.serverId}
activeConnection={active?.activeConnection ?? null}
daemonPublicKeyB64={active?.daemonPublicKeyB64}
client={client}
>
{null}
</SessionProvider>
@@ -158,6 +41,26 @@ function ManagedDaemonSession({ daemon }: { daemon: HostProfile }) {
export function MultiDaemonSessionHost() {
const { daemons } = useDaemonRegistry();
useEffect(() => {
const runtime = getHostRuntimeStore();
runtime.syncHosts(daemons);
}, [daemons]);
useEffect(() => {
const runtime = getHostRuntimeStore();
const subscription = AppState.addEventListener("change", (nextState) => {
if (nextState !== "active") {
return;
}
runtime.ensureConnectedAll();
void runtime.runProbeCycleNow();
});
return () => {
subscription.remove();
};
}, []);
if (daemons.length === 0) {
return null;
}
@@ -15,7 +15,7 @@ export type ConnectionState =
export type ConnectionStatus = ConnectionState["status"];
type ConnectionStateUpdate =
export type ConnectionStateUpdate =
| { status: "idle" }
| { status: "connecting"; activeConnection?: ActiveConnection | null; lastOnlineAt?: string | null }
| { status: "online"; activeConnection?: ActiveConnection | null; lastOnlineAt: string; agentListReady?: boolean }
+51 -51
View File
@@ -29,7 +29,6 @@ import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle";
import type { DaemonClient, ConnectionState } from "@server/client/daemon-client";
import { File } from "expo-file-system";
import { useDaemonConnections } from "./daemon-connections-context";
import type { ActiveConnection } from "./daemon-connections-context";
import {
useSessionStore,
type Agent,
@@ -204,24 +203,66 @@ const pushHistory = (history: string[], path: string): string[] => {
return [...normalizedHistory, path];
};
interface SessionProviderProps {
interface SessionProviderSharedProps {
children: ReactNode;
serverUrl: string;
serverId: string;
activeConnection: ActiveConnection | null;
}
interface SessionProviderUrlProps extends SessionProviderSharedProps {
serverUrl: string;
daemonPublicKeyB64?: string;
}
// SessionProvider: Daemon client message handler that updates Zustand store
export function SessionProvider({
interface SessionProviderClientProps extends SessionProviderSharedProps {
client: DaemonClient;
}
export type SessionProviderProps = SessionProviderUrlProps | SessionProviderClientProps;
function SessionProviderWithUrl({
children,
serverUrl,
serverId,
activeConnection,
serverUrl,
daemonPublicKeyB64,
}: SessionProviderProps) {
const queryClient = useQueryClient();
}: SessionProviderUrlProps) {
const client = useDaemonClient(serverUrl, { daemonPublicKeyB64 });
return (
<SessionProviderInternal
children={children}
serverId={serverId}
client={client}
/>
);
}
function SessionProviderWithClient({
children,
serverId,
client,
}: SessionProviderClientProps) {
return (
<SessionProviderInternal
children={children}
serverId={serverId}
client={client}
/>
);
}
// SessionProvider: Daemon client message handler that updates Zustand store
export function SessionProvider(props: SessionProviderProps) {
if ("client" in props) {
return <SessionProviderWithClient {...props} />;
}
return <SessionProviderWithUrl {...props} />;
}
function SessionProviderInternal({
children,
serverId,
client,
}: SessionProviderClientProps) {
const queryClient = useQueryClient();
const [connectionSnapshot, setConnectionSnapshot] =
useState<DaemonConnectionSnapshot>(() =>
mapConnectionState(client.getConnectionState(), client.lastError)
@@ -442,41 +483,6 @@ export function SessionProvider({
updateSessionClient(serverId, client);
}, [serverId, client, updateSessionClient]);
// Connection status tracking
useEffect(() => {
if (connectionSnapshot.isConnected) {
updateConnectionStatus(serverId, {
status: "online",
activeConnection,
lastOnlineAt: new Date().toISOString(),
});
return;
}
if (connectionSnapshot.isConnecting) {
updateConnectionStatus(serverId, { status: "connecting", activeConnection });
return;
}
if (connectionSnapshot.lastError) {
updateConnectionStatus(serverId, {
status: "error",
activeConnection,
lastError: connectionSnapshot.lastError,
});
return;
}
updateConnectionStatus(serverId, { status: "offline", activeConnection });
}, [
serverId,
updateConnectionStatus,
connectionSnapshot.isConnected,
connectionSnapshot.isConnecting,
connectionSnapshot.lastError,
activeConnection,
]);
// If the client drops mid-initialization, clear pending flags
useEffect(() => {
if (!connectionSnapshot.isConnected) {
@@ -616,12 +622,6 @@ export function SessionProvider({
]
);
useEffect(() => {
return () => {
updateConnectionStatus(serverId, { status: "offline", activeConnection, lastError: null });
};
}, [serverId, updateConnectionStatus, activeConnection]);
const updateExplorerState = useCallback(
(agentId: string, updater: (state: any) => any) => {
setFileExplorer(serverId, (prev) => {
@@ -0,0 +1,70 @@
import { describe, expect, it } from "vitest";
import type { HostRuntimeSnapshot } from "./host-runtime";
import { toDaemonConnectionUpdateFromRuntime } from "./host-runtime-bridge";
const DIRECT_CONNECTION = {
type: "direct" as const,
endpoint: "lan:6767",
display: "lan:6767",
};
function makeSnapshot(
input: Partial<HostRuntimeSnapshot>
): HostRuntimeSnapshot {
return {
serverId: "srv_test",
activeConnectionId: "direct:lan:6767",
activeConnection: DIRECT_CONNECTION,
connectionStatus: "idle",
lastError: null,
lastOnlineAt: null,
probeByConnectionId: new Map(),
clientGeneration: 0,
...input,
};
}
describe("toDaemonConnectionUpdateFromRuntime", () => {
it("maps online snapshots to online updates with the same active connection", () => {
const snapshot = makeSnapshot({
connectionStatus: "online",
lastOnlineAt: "2026-02-19T20:00:00.000Z",
});
const update = toDaemonConnectionUpdateFromRuntime(snapshot);
expect(update).toEqual({
status: "online",
activeConnection: DIRECT_CONNECTION,
lastOnlineAt: "2026-02-19T20:00:00.000Z",
});
});
it("maps error snapshots to error updates with the same error message", () => {
const snapshot = makeSnapshot({
connectionStatus: "error",
lastError: "transport closed",
lastOnlineAt: "2026-02-19T20:00:00.000Z",
});
const update = toDaemonConnectionUpdateFromRuntime(snapshot);
expect(update).toEqual({
status: "error",
activeConnection: DIRECT_CONNECTION,
lastError: "transport closed",
lastOnlineAt: "2026-02-19T20:00:00.000Z",
});
});
it("maps missing runtime snapshots to offline updates", () => {
const update = toDaemonConnectionUpdateFromRuntime(null);
expect(update).toEqual({
status: "offline",
activeConnection: null,
lastError: null,
lastOnlineAt: null,
});
});
});
@@ -0,0 +1,47 @@
import type { ConnectionStateUpdate } from "@/contexts/daemon-connections-context";
import type { HostRuntimeSnapshot } from "./host-runtime";
export function toDaemonConnectionUpdateFromRuntime(
snapshot: HostRuntimeSnapshot | null
): ConnectionStateUpdate {
if (!snapshot) {
return {
status: "offline",
activeConnection: null,
lastError: null,
lastOnlineAt: null,
};
}
const activeConnection = snapshot.activeConnection;
switch (snapshot.connectionStatus) {
case "idle":
return { status: "idle" };
case "connecting":
return {
status: "connecting",
activeConnection,
lastOnlineAt: snapshot.lastOnlineAt,
};
case "online":
return {
status: "online",
activeConnection,
lastOnlineAt: snapshot.lastOnlineAt ?? new Date().toISOString(),
};
case "offline":
return {
status: "offline",
activeConnection,
lastError: null,
lastOnlineAt: snapshot.lastOnlineAt,
};
case "error":
return {
status: "error",
activeConnection,
lastError: snapshot.lastError ?? "Connection error",
lastOnlineAt: snapshot.lastOnlineAt,
};
}
}
@@ -0,0 +1,253 @@
import { describe, expect, it } from "vitest";
import type { DaemonClient, ConnectionState } from "@server/client/daemon-client";
import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context";
import {
HostRuntimeController,
type HostRuntimeControllerDeps,
} from "./host-runtime";
class FakeDaemonClient {
private state: ConnectionState = { status: "idle" };
private listeners = new Set<(status: ConnectionState) => void>();
private error: string | null = null;
public connectCalls = 0;
public closeCalls = 0;
public ensureConnectedCalls = 0;
async connect(): Promise<void> {
this.connectCalls += 1;
this.setConnectionState({ status: "connected" });
}
async close(): Promise<void> {
this.closeCalls += 1;
this.setConnectionState({ status: "disconnected", reason: "client_closed" });
}
ensureConnected(): void {
this.ensureConnectedCalls += 1;
if (this.state.status !== "connected") {
this.setConnectionState({ status: "connected" });
}
}
getConnectionState(): ConnectionState {
return this.state;
}
subscribeConnectionStatus(
listener: (status: ConnectionState) => void
): () => void {
this.listeners.add(listener);
listener(this.state);
return () => {
this.listeners.delete(listener);
};
}
get lastError(): string | null {
return this.error;
}
setConnectionState(next: ConnectionState): void {
this.state = next;
if (next.status === "disconnected") {
this.error = next.reason ?? this.error;
}
for (const listener of this.listeners) {
listener(next);
}
}
}
function makeHost(input?: Partial<HostProfile>): HostProfile {
const direct: HostConnection = {
id: "direct:lan:6767",
type: "direct",
endpoint: "lan:6767",
};
const relay: HostConnection = {
id: "relay:relay.paseo.sh:443",
type: "relay",
relayEndpoint: "relay.paseo.sh:443",
daemonPublicKeyB64: "pk_test",
};
return {
serverId: input?.serverId ?? "srv_test",
label: input?.label ?? "test host",
connections: input?.connections ?? [direct, relay],
preferredConnectionId: input?.preferredConnectionId ?? direct.id,
createdAt: input?.createdAt ?? new Date(0).toISOString(),
updatedAt: input?.updatedAt ?? new Date(0).toISOString(),
};
}
function makeDeps(
latencyByConnectionId: Record<string, number | Error>,
createdClients: FakeDaemonClient[]
): HostRuntimeControllerDeps {
return {
createClient: () => {
const client = new FakeDaemonClient();
createdClients.push(client);
return client as unknown as DaemonClient;
},
measureLatency: async ({ connection }) => {
const value = latencyByConnectionId[connection.id];
if (value instanceof Error) {
throw value;
}
if (typeof value !== "number") {
throw new Error(`missing latency for ${connection.id}`);
}
return value;
},
};
}
describe("HostRuntimeController", () => {
it("selects the lowest-latency connection on startup", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 82,
"relay:relay.paseo.sh:443": 18,
};
const controller = new HostRuntimeController({
host,
deps: makeDeps(latencies, clients),
});
await controller.start({ autoProbe: false });
const snapshot = controller.getSnapshot();
expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(snapshot.connectionStatus).toBe("online");
expect(clients).toHaveLength(1);
expect(clients[0]?.connectCalls).toBe(1);
});
it("fails over when active connection becomes unavailable", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 15,
"relay:relay.paseo.sh:443": 55,
};
const controller = new HostRuntimeController({
host,
deps: makeDeps(latencies, clients),
});
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
expect(clients).toHaveLength(1);
latencies["direct:lan:6767"] = new Error("direct unavailable");
latencies["relay:relay.paseo.sh:443"] = 42;
await controller.runProbeCycleNow();
const snapshot = controller.getSnapshot();
expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(snapshot.connectionStatus).toBe("online");
expect(clients).toHaveLength(2);
expect(clients[0]?.closeCalls).toBe(1);
});
it("switches only after the faster alternative wins consecutive probes", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 15,
"relay:relay.paseo.sh:443": 60,
};
const controller = new HostRuntimeController({
host,
deps: makeDeps(latencies, clients),
});
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 95;
latencies["relay:relay.paseo.sh:443"] = 30;
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(clients).toHaveLength(2);
});
it("does not switch on a transient latency spike", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 15,
"relay:relay.paseo.sh:443": 80,
};
const controller = new HostRuntimeController({
host,
deps: makeDeps(latencies, clients),
});
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 100;
latencies["relay:relay.paseo.sh:443"] = 20;
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 20;
latencies["relay:relay.paseo.sh:443"] = 90;
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 100;
latencies["relay:relay.paseo.sh:443"] = 20;
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("relay:relay.paseo.sh:443");
});
it("exposes one snapshot with active connection and status from same source", async () => {
const host = makeHost();
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 12,
"relay:relay.paseo.sh:443": 65,
};
const controller = new HostRuntimeController({
host,
deps: makeDeps(latencies, clients),
});
const observed = new Array<ReturnType<typeof controller.getSnapshot>>();
const unsubscribe = controller.subscribe(() => {
observed.push(controller.getSnapshot());
});
await controller.start({ autoProbe: false });
clients[0]?.setConnectionState({
status: "disconnected",
reason: "transport closed",
});
const latest = observed[observed.length - 1];
expect(latest?.activeConnectionId).toBe("direct:lan:6767");
expect(latest?.connectionStatus).toBe("error");
expect(latest?.lastError).toBe("transport closed");
unsubscribe();
});
});
+577
View File
@@ -0,0 +1,577 @@
import { useMemo, useSyncExternalStore } from "react";
import { DaemonClient, type ConnectionState } from "@server/client/daemon-client";
import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context";
import type { ActiveConnection } from "@/contexts/daemon-connections-context";
import {
buildDaemonWebSocketUrl,
buildRelayWebSocketUrl,
} from "@/utils/daemon-endpoints";
import { measureConnectionLatency } from "@/utils/test-daemon-connection";
import {
selectBestConnection,
type ConnectionCandidate,
type ConnectionProbeState,
} from "@/utils/connection-selection";
import { createTauriWebSocketTransportFactory } from "@/utils/tauri-daemon-transport";
export type HostRuntimeConnectionStatus =
| "idle"
| "connecting"
| "online"
| "offline"
| "error";
export type HostRuntimeSnapshot = {
serverId: string;
activeConnectionId: string | null;
activeConnection: ActiveConnection | null;
connectionStatus: HostRuntimeConnectionStatus;
lastError: string | null;
lastOnlineAt: string | null;
probeByConnectionId: Map<string, ConnectionProbeState>;
clientGeneration: number;
};
export type HostRuntimeControllerDeps = {
createClient: (input: {
host: HostProfile;
connection: HostConnection;
}) => DaemonClient;
measureLatency: (input: {
host: HostProfile;
connection: HostConnection;
}) => Promise<number>;
};
export type HostRuntimeStartOptions = {
autoProbe?: boolean;
};
const PROBE_INTERVAL_MS = 10_000;
const ADAPTIVE_SWITCH_THRESHOLD_MS = 40;
const ADAPTIVE_SWITCH_CONSECUTIVE_PROBES = 3;
function toActiveConnection(connection: HostConnection): ActiveConnection {
if (connection.type === "direct") {
return {
type: "direct",
endpoint: connection.endpoint,
display: connection.endpoint,
};
}
return {
type: "relay",
endpoint: connection.relayEndpoint,
display: "relay",
};
}
function mapClientConnectionState(input: {
state: ConnectionState;
lastError: string | null;
}): Pick<HostRuntimeSnapshot, "connectionStatus" | "lastError" | "lastOnlineAt"> {
const { state, lastError } = input;
if (state.status === "connected") {
return {
connectionStatus: "online",
lastError: null,
lastOnlineAt: new Date().toISOString(),
};
}
if (state.status === "connecting") {
return {
connectionStatus: "connecting",
lastError: null,
lastOnlineAt: null,
};
}
if (state.status === "idle") {
return {
connectionStatus: "idle",
lastError: null,
lastOnlineAt: null,
};
}
const reason = state.reason ?? lastError ?? null;
if (!reason || reason === "client_closed") {
return {
connectionStatus: "offline",
lastError: null,
lastOnlineAt: null,
};
}
return {
connectionStatus: "error",
lastError: reason,
lastOnlineAt: null,
};
}
function buildConnectionCandidates(host: HostProfile): ConnectionCandidate[] {
return host.connections.map((connection) => ({
connectionId: connection.id,
connection,
}));
}
function findConnectionById(
host: HostProfile,
connectionId: string | null
): HostConnection | null {
if (!connectionId) {
return null;
}
return host.connections.find((connection) => connection.id === connectionId) ?? null;
}
function selectInitialConnectionId(host: HostProfile): string | null {
if (host.connections.length === 0) {
return null;
}
return host.connections[0]?.id ?? null;
}
function createDefaultDeps(): HostRuntimeControllerDeps {
return {
createClient: ({ host, connection }) => {
const tauriTransportFactory = createTauriWebSocketTransportFactory();
const base = {
suppressSendErrors: true,
...(tauriTransportFactory
? { transportFactory: tauriTransportFactory }
: {}),
};
if (connection.type === "direct") {
return new DaemonClient({
...base,
url: buildDaemonWebSocketUrl(connection.endpoint),
});
}
return new DaemonClient({
...base,
url: buildRelayWebSocketUrl({
endpoint: connection.relayEndpoint,
serverId: host.serverId,
}),
e2ee: {
enabled: true,
daemonPublicKeyB64: connection.daemonPublicKeyB64,
},
});
},
measureLatency: ({ host, connection }) =>
measureConnectionLatency(connection, { serverId: host.serverId }),
};
}
export class HostRuntimeController {
private host: HostProfile;
private deps: HostRuntimeControllerDeps;
private snapshot: HostRuntimeSnapshot;
private listeners = new Set<() => void>();
private activeClient: DaemonClient | null = null;
private unsubscribeClientStatus: (() => void) | null = null;
private probeIntervalHandle: ReturnType<typeof setInterval> | null = null;
private started = false;
private switchCandidateConnectionId: string | null = null;
private switchCandidateHitCount = 0;
constructor(input: {
host: HostProfile;
deps?: HostRuntimeControllerDeps;
}) {
this.host = input.host;
this.deps = input.deps ?? createDefaultDeps();
this.snapshot = {
serverId: this.host.serverId,
activeConnectionId: null,
activeConnection: null,
connectionStatus: "idle",
lastError: null,
lastOnlineAt: null,
probeByConnectionId: new Map(),
clientGeneration: 0,
};
}
getSnapshot(): HostRuntimeSnapshot {
return this.snapshot;
}
getClient(): DaemonClient | null {
return this.activeClient;
}
subscribe(listener: () => void): () => void {
this.listeners.add(listener);
return () => {
this.listeners.delete(listener);
};
}
async start(options?: HostRuntimeStartOptions): Promise<void> {
if (this.started) {
return;
}
this.started = true;
await this.runProbeCycleNow();
if (!this.snapshot.activeConnectionId) {
const fallbackConnectionId = selectInitialConnectionId(this.host);
if (fallbackConnectionId) {
await this.switchToConnection(fallbackConnectionId);
}
}
if (options?.autoProbe !== false && PROBE_INTERVAL_MS > 0) {
this.probeIntervalHandle = setInterval(() => {
void this.runProbeCycleNow();
}, PROBE_INTERVAL_MS);
}
}
async stop(): Promise<void> {
this.started = false;
if (this.probeIntervalHandle) {
clearInterval(this.probeIntervalHandle);
this.probeIntervalHandle = null;
}
if (this.unsubscribeClientStatus) {
this.unsubscribeClientStatus();
this.unsubscribeClientStatus = null;
}
if (this.activeClient) {
const prev = this.activeClient;
this.activeClient = null;
await prev.close().catch(() => undefined);
}
this.updateSnapshot({
activeConnectionId: null,
activeConnection: null,
connectionStatus: "offline",
lastError: null,
lastOnlineAt: null,
});
}
async updateHost(host: HostProfile): Promise<void> {
this.host = host;
const activeConnectionId = this.snapshot.activeConnectionId;
const activeConnection = findConnectionById(host, activeConnectionId);
if (!activeConnection) {
await this.runProbeCycleNow();
if (this.snapshot.activeConnectionId) {
return;
}
const fallbackConnectionId = selectInitialConnectionId(host);
if (!fallbackConnectionId) {
await this.stop();
return;
}
await this.switchToConnection(fallbackConnectionId);
}
}
ensureConnected(): void {
this.activeClient?.ensureConnected();
}
async runProbeCycleNow(): Promise<void> {
if (this.host.connections.length === 0) {
this.updateSnapshot({
probeByConnectionId: new Map(),
});
return;
}
const probeByConnectionId = new Map<string, ConnectionProbeState>();
await Promise.all(
this.host.connections.map(async (connection) => {
try {
const latencyMs = await this.deps.measureLatency({
host: this.host,
connection,
});
probeByConnectionId.set(connection.id, {
status: "available",
latencyMs,
});
} catch {
probeByConnectionId.set(connection.id, {
status: "unavailable",
latencyMs: null,
});
}
})
);
this.updateSnapshot({ probeByConnectionId });
const activeConnectionId = this.snapshot.activeConnectionId;
const activeProbe = activeConnectionId
? probeByConnectionId.get(activeConnectionId)
: null;
if (!activeConnectionId || !findConnectionById(this.host, activeConnectionId)) {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId) {
await this.switchToConnection(nextConnectionId);
}
return;
}
if (activeProbe?.status === "unavailable") {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId && nextConnectionId !== activeConnectionId) {
await this.switchToConnection(nextConnectionId);
}
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
return;
}
if (activeProbe && activeProbe.status === "available") {
const available = Array.from(probeByConnectionId.entries())
.filter(([, probe]) => probe.status === "available")
.map(([connectionId, probe]) => ({
connectionId,
latencyMs: (probe as { status: "available"; latencyMs: number }).latencyMs,
}))
.sort((left, right) => left.latencyMs - right.latencyMs);
const fastest = available[0] ?? null;
if (!fastest || fastest.connectionId === activeConnectionId) {
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
return;
}
const activeLatency = activeProbe.latencyMs;
const improvement = activeLatency - fastest.latencyMs;
if (improvement < ADAPTIVE_SWITCH_THRESHOLD_MS) {
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
return;
}
if (this.switchCandidateConnectionId === fastest.connectionId) {
this.switchCandidateHitCount += 1;
} else {
this.switchCandidateConnectionId = fastest.connectionId;
this.switchCandidateHitCount = 1;
}
if (
this.switchCandidateHitCount >=
ADAPTIVE_SWITCH_CONSECUTIVE_PROBES
) {
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
await this.switchToConnection(fastest.connectionId);
}
}
}
private updateSnapshot(
patch: Partial<Omit<HostRuntimeSnapshot, "serverId" | "clientGeneration">>
): void {
const next: HostRuntimeSnapshot = {
...this.snapshot,
...patch,
serverId: this.host.serverId,
};
this.snapshot = next;
for (const listener of this.listeners) {
listener();
}
}
private async switchToConnection(connectionId: string): Promise<void> {
const connection = findConnectionById(this.host, connectionId);
if (!connection) {
return;
}
if (this.unsubscribeClientStatus) {
this.unsubscribeClientStatus();
this.unsubscribeClientStatus = null;
}
if (this.activeClient) {
const previousClient = this.activeClient;
this.activeClient = null;
await previousClient.close().catch(() => undefined);
}
const client = this.deps.createClient({
host: this.host,
connection,
});
this.activeClient = client;
this.snapshot = {
...this.snapshot,
serverId: this.host.serverId,
activeConnectionId: connection.id,
activeConnection: toActiveConnection(connection),
connectionStatus: "connecting",
lastError: null,
lastOnlineAt: null,
clientGeneration: this.snapshot.clientGeneration + 1,
};
for (const listener of this.listeners) {
listener();
}
this.unsubscribeClientStatus = client.subscribeConnectionStatus((state) => {
const mapped = mapClientConnectionState({
state,
lastError: client.lastError,
});
this.updateSnapshot(mapped);
});
try {
await client.connect();
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
this.updateSnapshot({
connectionStatus: "error",
lastError: message,
});
}
}
}
export class HostRuntimeStore {
private controllers = new Map<string, HostRuntimeController>();
private serverListeners = new Map<string, Set<() => void>>();
private deps: HostRuntimeControllerDeps;
constructor(input?: {
deps?: HostRuntimeControllerDeps;
}) {
this.deps = input?.deps ?? createDefaultDeps();
}
syncHosts(hosts: HostProfile[]): void {
const nextIds = new Set(hosts.map((host) => host.serverId));
for (const [serverId, controller] of this.controllers) {
if (nextIds.has(serverId)) {
continue;
}
this.controllers.delete(serverId);
void controller.stop();
this.emit(serverId);
}
for (const host of hosts) {
const existing = this.controllers.get(host.serverId);
if (existing) {
void existing.updateHost(host);
continue;
}
const controller = new HostRuntimeController({
host,
deps: this.deps,
});
this.controllers.set(host.serverId, controller);
controller.subscribe(() => {
this.emit(host.serverId);
});
void controller.start();
this.emit(host.serverId);
}
}
getSnapshot(serverId: string): HostRuntimeSnapshot | null {
return this.controllers.get(serverId)?.getSnapshot() ?? null;
}
getClient(serverId: string): DaemonClient | null {
return this.controllers.get(serverId)?.getClient() ?? null;
}
subscribe(serverId: string, listener: () => void): () => void {
const existing = this.serverListeners.get(serverId) ?? new Set<() => void>();
existing.add(listener);
this.serverListeners.set(serverId, existing);
return () => {
const set = this.serverListeners.get(serverId);
if (!set) {
return;
}
set.delete(listener);
if (set.size === 0) {
this.serverListeners.delete(serverId);
}
};
}
ensureConnectedAll(): void {
for (const controller of this.controllers.values()) {
controller.ensureConnected();
}
}
runProbeCycleNow(serverId?: string): Promise<void> {
if (serverId) {
return this.controllers.get(serverId)?.runProbeCycleNow() ?? Promise.resolve();
}
return Promise.all(
Array.from(this.controllers.values(), (controller) =>
controller.runProbeCycleNow()
)
).then(() => undefined);
}
private emit(serverId: string): void {
const listeners = this.serverListeners.get(serverId);
if (!listeners) {
return;
}
for (const listener of listeners) {
listener();
}
}
}
let singletonHostRuntimeStore: HostRuntimeStore | null = null;
export function getHostRuntimeStore(): HostRuntimeStore {
if (!singletonHostRuntimeStore) {
singletonHostRuntimeStore = new HostRuntimeStore();
}
return singletonHostRuntimeStore;
}
export function useHostRuntimeSnapshot(
serverId: string
): HostRuntimeSnapshot | null {
const store = getHostRuntimeStore();
return useSyncExternalStore(
(onStoreChange) => store.subscribe(serverId, onStoreChange),
() => store.getSnapshot(serverId),
() => store.getSnapshot(serverId)
);
}
export function useHostRuntimeSession(serverId: string): {
snapshot: HostRuntimeSnapshot | null;
client: DaemonClient | null;
} {
const store = getHostRuntimeStore();
const snapshot = useHostRuntimeSnapshot(serverId);
return useMemo(
() => ({
snapshot,
client: store.getClient(serverId),
}),
[serverId, snapshot, store]
);
}
@@ -25,7 +25,7 @@ function probes(
}
describe("selectBestConnection", () => {
it("picks the available connection with lowest latency regardless of transport type", () => {
it("picks the available connection with lowest latency regardless of transport", () => {
const candidates: ConnectionCandidate[] = [
{ connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") },
{
@@ -36,7 +36,6 @@ describe("selectBestConnection", () => {
const selected = selectBestConnection({
candidates,
preferredConnectionId: "direct:a",
probeByConnectionId: probes({
"direct:a": { status: "available", latencyMs: 84 },
"relay:b": { status: "available", latencyMs: 34 },
@@ -46,6 +45,28 @@ describe("selectBestConnection", () => {
expect(selected).toBe("relay:b");
});
it("picks the lowest-latency connection among mixed transport candidates", () => {
const candidates: ConnectionCandidate[] = [
{ connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") },
{ connectionId: "direct:c", connection: makeDirect("direct:c", "c:6767") },
{
connectionId: "relay:b",
connection: makeRelay("relay:b", "relay.example:443"),
},
];
const selected = selectBestConnection({
candidates,
probeByConnectionId: probes({
"direct:a": { status: "available", latencyMs: 84 },
"direct:c": { status: "available", latencyMs: 41 },
"relay:b": { status: "available", latencyMs: 12 },
}),
});
expect(selected).toBe("relay:b");
});
it("ignores unavailable and pending probes", () => {
const candidates: ConnectionCandidate[] = [
{ connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") },
@@ -58,7 +79,6 @@ describe("selectBestConnection", () => {
const selected = selectBestConnection({
candidates,
preferredConnectionId: "direct:a",
probeByConnectionId: probes({
"direct:a": { status: "pending", latencyMs: null },
"relay:b": { status: "unavailable", latencyMs: null },
@@ -69,7 +89,7 @@ describe("selectBestConnection", () => {
expect(selected).toBe("direct:c");
});
it("falls back to preferred connection when no candidates are available", () => {
it("returns null when no candidates are available", () => {
const candidates: ConnectionCandidate[] = [
{ connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") },
{
@@ -80,17 +100,16 @@ describe("selectBestConnection", () => {
const selected = selectBestConnection({
candidates,
preferredConnectionId: "relay:b",
probeByConnectionId: probes({
"direct:a": { status: "pending", latencyMs: null },
"relay:b": { status: "unavailable", latencyMs: null },
}),
});
expect(selected).toBe("relay:b");
expect(selected).toBeNull();
});
it("falls back to first candidate when preferred is missing and none are available", () => {
it("returns null when no probes are available even if candidates exist", () => {
const candidates: ConnectionCandidate[] = [
{ connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") },
{
@@ -101,13 +120,12 @@ describe("selectBestConnection", () => {
const selected = selectBestConnection({
candidates,
preferredConnectionId: "missing",
probeByConnectionId: probes({
"direct:a": { status: "unavailable", latencyMs: null },
"relay:b": { status: "pending", latencyMs: null },
}),
});
expect(selected).toBe("direct:a");
expect(selected).toBeNull();
});
});
+9 -15
View File
@@ -12,14 +12,13 @@ export type ConnectionProbeState =
export type SelectBestConnectionInput = {
candidates: ConnectionCandidate[];
preferredConnectionId: string | null;
probeByConnectionId: Map<string, ConnectionProbeState>;
};
export function selectBestConnection(
input: SelectBestConnectionInput
): string | null {
const { candidates, preferredConnectionId, probeByConnectionId } = input;
const { candidates, probeByConnectionId } = input;
if (candidates.length === 0) {
return null;
}
@@ -29,8 +28,9 @@ export function selectBestConnection(
.filter(({ candidate }) => {
const probe = probeByConnectionId.get(candidate.connectionId);
return probe?.status === "available";
})
.sort((left, right) => {
});
const byLatency = (left: { candidate: ConnectionCandidate; index: number }, right: { candidate: ConnectionCandidate; index: number }) => {
const leftProbe = probeByConnectionId.get(left.candidate.connectionId);
const rightProbe = probeByConnectionId.get(right.candidate.connectionId);
@@ -48,18 +48,12 @@ export function selectBestConnection(
}
return leftLatency - rightLatency;
});
};
if (available.length > 0) {
return available[0]!.candidate.connectionId;
if (available.length === 0) {
return null;
}
const preferred = candidates.find(
(candidate) => candidate.connectionId === preferredConnectionId
);
if (preferred) {
return preferred.connectionId;
}
return candidates[0]!.connectionId;
const sorted = available.sort(byLatency);
return sorted[0]!.candidate.connectionId;
}