jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(app): stabilize startup connection routing

Browse Source
This commit is contained in:
Mohamed Boudra
2026-04-22 13:11:41 +07:00
parent 9a39ba4be0
commit 6b210df471
7 changed files with 187 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/runtime/host-runtime.test.ts
+110 -28
View File
@@ -213,17 +213,21 @@ function makeDeps(
return client as unknown as DaemonClient;
},
connectToDaemon: async ({ host, connection }) => {
const value = latencyByConnectionId[connection.id];
if (value instanceof Error) {
throw value;
}
if (typeof value !== "number") {
throw new Error(`missing latency for ${connection.id}`);
}
const readLatency = (): number => {
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;
};
readLatency();
const client = new FakeDaemonClient();
client.connectCalls = 1;
client.setConnectionState({ status: "connected" });
client.ping = async () => ({ rttMs: value });
client.ping = async () => ({ rttMs: readLatency() });
createdClients.push(client);
return {
client: client as unknown as DaemonClient,
@@ -433,7 +437,95 @@ describe("HostRuntimeController", () => {
await probeCycle;
});
it("fails over when active connection becomes unavailable", async () => {
it("probes the active online connection through the existing client", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const probeAttempts: string[] = [];
const latencies: Record<string, number | Error> = {
"direct:lan:6767": 12,
"relay:relay.paseo.sh:443": 65,
};
const controller = new HostRuntimeController({
host,
deps: {
createClient: () => {
throw new Error("should adopt probe clients");
},
connectToDaemon: async ({ host, connection }) => {
probeAttempts.push(connection.id);
const value = latencies[connection.id];
if (value instanceof Error) {
throw value;
}
if (typeof value !== "number") {
throw new Error(`missing latency for ${connection.id}`);
}
return {
client: makeConnectedProbeClient(value) as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
};
},
getClientId: async () => "cid_test_runtime",
},
});
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
probeAttempts.length = 0;
const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient;
activeClient.ping = async () => ({ rttMs: 9 });
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(probeAttempts).toEqual(["relay:relay.paseo.sh:443"]);
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
expect(controller.getSnapshot().connectionStatus).toBe("online");
expect(controller.getSnapshot().probeByConnectionId.get("direct:lan:6767")).toEqual({
status: "available",
latencyMs: 9,
});
});
it("rejects probes that resolve to a different server id", async () => {
const host = makeHost({
serverId: "srv_old",
connections: [
{
id: "direct:localhost:6767",
type: "directTcp",
endpoint: "localhost:6767",
},
],
});
const mismatchedClient = makeConnectedProbeClient(8);
const controller = new HostRuntimeController({
host,
deps: {
createClient: () => {
throw new Error("should not create active client");
},
connectToDaemon: async () => ({
client: mismatchedClient as unknown as DaemonClient,
serverId: "srv_current",
hostname: "current host",
}),
getClientId: async () => "cid_test_runtime",
},
});
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().connectionStatus).toBe("connecting");
expect(controller.getSnapshot().activeConnectionId).toBeNull();
expect(controller.getSnapshot().probeByConnectionId.get("direct:localhost:6767")).toEqual({
status: "unavailable",
latencyMs: null,
});
expect(mismatchedClient.closeCalls).toBe(1);
});
it("fails over when the active client ping fails", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
@@ -450,6 +542,9 @@ describe("HostRuntimeController", () => {
const initialClient = controller.getSnapshot().client;
expect(initialClient).toBeTruthy();
(initialClient as unknown as FakeDaemonClient).ping = async () => {
throw new Error("active ping failed");
};
latencies["direct:lan:6767"] = new Error("direct unavailable");
latencies["relay:relay.paseo.sh:443"] = 42;
clearProbeBackoff(controller);
@@ -838,7 +933,7 @@ describe("HostRuntimeController", () => {
expect(createdClients[0]?.closeCalls).toBe(1);
});
it("ignores stale probe results when overlapping probe cycles finish out of order", async () => {
it("coalesces overlapping probe cycles instead of invalidating the in-flight result", async () => {
const host = makeHost({
connections: [
{
@@ -849,7 +944,6 @@ describe("HostRuntimeController", () => {
],
});
const slowProbe = createDeferred<number>();
const fastProbe = createDeferred<number>();
let probeCalls = 0;
const controller = new HostRuntimeController({
@@ -865,9 +959,6 @@ describe("HostRuntimeController", () => {
if (probeCalls === 1) {
return { rttMs: await slowProbe.promise };
}
if (probeCalls === 2) {
return { rttMs: await fastProbe.promise };
}
throw new Error("unexpected probe call");
};
return {
@@ -883,23 +974,14 @@ describe("HostRuntimeController", () => {
const first = controller.runProbeCycleNow();
clearProbeBackoff(controller);
const second = controller.runProbeCycleNow();
fastProbe.resolve(12);
await second;
const probeAfterSecond = controller.getSnapshot().probeByConnectionId.get("direct:lan:6767");
expect(probeAfterSecond).toEqual({
status: "available",
latencyMs: 12,
});
expect(probeCalls).toBe(1);
slowProbe.resolve(900);
await first;
const probeAfterFirstSettles = controller
.getSnapshot()
.probeByConnectionId.get("direct:lan:6767");
expect(probeAfterFirstSettles).toEqual({
await Promise.all([first, second]);
const probeAfterCycle = controller.getSnapshot().probeByConnectionId.get("direct:lan:6767");
expect(probeAfterCycle).toEqual({
status: "available",
latencyMs: 12,
latencyMs: 900,
});
});
packages/app/src/runtime/host-runtime.ts
+51 -27
View File
@@ -11,7 +11,6 @@ import {
connectionFromListen,
normalizeStoredHostProfile,
upsertHostConnectionInProfiles,
registryHasDirectEndpoint,
type HostConnection,
type HostProfile,
} from "@/types/host-connection";
@@ -489,6 +488,7 @@ export class HostRuntimeController {
private clientIdHash: string | null = null;
private switchRequestVersion = 0;
private probeRequestVersion = 0;
private probeCycleInFlight: Promise<void> | null = null;
constructor(input: {
host: HostProfile;
@@ -629,6 +629,20 @@ export class HostRuntimeController {
}
async runProbeCycleNow(): Promise<void> {
if (this.probeCycleInFlight) {
return this.probeCycleInFlight;
}
const cycle = this.runProbeCycle().finally(() => {
if (this.probeCycleInFlight === cycle) {
this.probeCycleInFlight = null;
}
});
this.probeCycleInFlight = cycle;
return cycle;
}
private async runProbeCycle(): Promise<void> {
const requestVersion = ++this.probeRequestVersion;
if (this.host.connections.length === 0) {
if (!this.isCurrentProbeRequest(requestVersion)) {
@@ -664,10 +678,15 @@ export class HostRuntimeController {
const probeByConnectionId = new Map(this.snapshot.probeByConnectionId);
for (const connection of connectionsToProbe) {
this.connectionLastProbedAt.set(connection.id, performance.now());
probeByConnectionId.set(connection.id, {
status: "pending",
latencyMs: null,
});
const existingProbe = probeByConnectionId.get(connection.id);
const shouldPreserveActiveLatency =
isOnline && connection.id === activeConnectionId && existingProbe?.status === "available";
if (!shouldPreserveActiveLatency) {
probeByConnectionId.set(connection.id, {
status: "pending",
latencyMs: null,
});
}
}
this.updateSnapshot({ probeByConnectionId: new Map(probeByConnectionId) });
@@ -801,22 +820,39 @@ export class HostRuntimeController {
for (const connection of connectionsToProbe) {
void (async () => {
let connectedClient: DaemonClient | null = null;
let handedOffClient = false;
let shouldCloseClient = false;
try {
const { client } = await this.deps.connectToDaemon({
host: this.host,
connection,
});
connectedClient = client;
const activeClient =
this.snapshot.connectionStatus === "online" &&
this.snapshot.activeConnectionId === connection.id
? this.snapshot.client
: null;
if (activeClient) {
connectedClient = activeClient;
} else {
const { client, serverId } = await this.deps.connectToDaemon({
host: this.host,
connection,
});
if (serverId !== this.host.serverId) {
await client.close().catch(() => undefined);
throw new Error(
`Connection resolved to ${serverId}, expected ${this.host.serverId}.`,
);
}
connectedClient = client;
shouldCloseClient = true;
}
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
const activated = await maybeActivateFirstAvailable(connection.id, client);
handedOffClient = activated;
const activated = await maybeActivateFirstAvailable(connection.id, connectedClient);
shouldCloseClient = shouldCloseClient && !activated;
const { rttMs } = await client.ping({ timeoutMs: 5000 });
const { rttMs } = await connectedClient.ping({ timeoutMs: 5000 });
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
@@ -835,7 +871,7 @@ export class HostRuntimeController {
publishProbeState();
}
} finally {
if (connectedClient && !handedOffClient) {
if (connectedClient && shouldCloseClient) {
await connectedClient.close().catch(() => undefined);
}
settleProbe();
@@ -1100,7 +1136,6 @@ export class HostRuntimeController {
const REGISTRY_STORAGE_KEY = "@paseo:daemon-registry";
const DEFAULT_LOCALHOST_ENDPOINT = process.env.EXPO_PUBLIC_LOCAL_DAEMON?.trim() || "localhost:6767";
const DEFAULT_LOCALHOST_BOOTSTRAP_KEY = "@paseo:default-localhost-bootstrap-v1";
const DEFAULT_LOCALHOST_BOOTSTRAP_TIMEOUT_MS = 2500;
const CONNECTION_ONLINE_TIMEOUT_MS = 15_000;
const E2E_STORAGE_KEY = "@paseo:e2e";
@@ -1230,16 +1265,6 @@ export class HostRuntimeStore {
private async bootstrapLocalhost(): Promise<void> {
try {
const alreadyHandled = await AsyncStorage.getItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY);
if (alreadyHandled) {
return;
}
if (registryHasDirectEndpoint(this.hosts, DEFAULT_LOCALHOST_ENDPOINT)) {
await AsyncStorage.setItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY, "1");
return;
}
try {
const { client, serverId, hostname } = await connectToDaemon(
{
@@ -1256,7 +1281,6 @@ export class HostRuntimeStore {
label: hostname ?? undefined,
existingClient: client,
});
await AsyncStorage.setItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY, "1");
} catch {
// Best-effort bootstrap only
}