diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 5a8e5bb7e..b76e85fdf 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { DaemonClient, ConnectionState, @@ -12,16 +12,16 @@ import { HostRuntimeController, HostRuntimeStore, type HostRuntimeControllerDeps, - type HostRuntimeSnapshot, } from "./host-runtime"; class FakeDaemonClient { private state: ConnectionState = { status: "idle" }; private listeners = new Set<(status: ConnectionState) => void>(); private error: string | null = null; + private heartbeatRttMs: number | null = null; + private latencyMeasurementFailure: Error | null = null; + private latencyMeasurementsRequested: Array<{ timeoutMs?: number }> = []; public connectCalls = 0; - public closeCalls = 0; - public ensureConnectedCalls = 0; public fetchAgentsCalls: FetchAgentsOptions[] = []; public fetchAgentsResponses: Awaited>[] = []; @@ -31,12 +31,10 @@ class FakeDaemonClient { } async close(): Promise { - this.closeCalls += 1; this.setConnectionState({ status: "disconnected", reason: "client_closed" }); } ensureConnected(): void { - this.ensureConnectedCalls += 1; if (this.state.status !== "connected") { this.setConnectionState({ status: "connected" }); } @@ -76,12 +74,41 @@ class FakeDaemonClient { return { rttMs: 0 }; } - async checkLiveness(): Promise<{ rttMs: number }> { - return this.ping(); + async measureLatency(params?: { timeoutMs?: number }): Promise { + this.latencyMeasurementsRequested.push(params ?? {}); + if (this.latencyMeasurementFailure) { + throw this.latencyMeasurementFailure; + } + const result = await this.ping(); + return result.rttMs; } setReconnectEnabled(_enabled: boolean): void {} + getLastLivenessRttMs(): number | null { + return this.heartbeatRttMs; + } + + heartbeatReportsRtt(rttMs: number | null): void { + this.heartbeatRttMs = rttMs; + } + + latencyMeasurementsFailWith(message: string): void { + this.latencyMeasurementFailure = new Error(message); + } + + latencyMeasurements(): Array<{ timeoutMs?: number }> { + return this.latencyMeasurementsRequested; + } + + clearLatencyMeasurements(): void { + this.latencyMeasurementsRequested = []; + } + + isDisposed(): boolean { + return this.state.status === "disconnected" && this.state.reason === "client_closed"; + } + setConnectionState(next: ConnectionState): void { this.state = next; if (next.status === "disconnected") { @@ -93,6 +120,16 @@ class FakeDaemonClient { } } +afterEach(() => { + vi.useRealTimers(); +}); + +function useHostRuntimeClock(): void { + vi.useFakeTimers({ + toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval", "performance"], + }); +} + function makeFetchAgentsPayload(input: { entries: FetchAgentsEntry[]; hasMore?: boolean; @@ -275,36 +312,6 @@ function makeConnectedProbeClient(latencyMs: number): FakeDaemonClient { return client; } -function clearProbeBackoff(controller: HostRuntimeController): void { - ( - controller as unknown as { - connectionLastProbedAt: Map; - } - ).connectionLastProbedAt.clear(); -} - -type HostRuntimeSnapshotPatch = Partial>; - -function updateControllerSnapshot( - controller: HostRuntimeController, - patch: HostRuntimeSnapshotPatch, -): void { - ( - controller as unknown as { - updateSnapshot: (patch: HostRuntimeSnapshotPatch) => void; - } - ).updateSnapshot(patch); -} - -function makeProbeMap( - entries: [ - string, - HostRuntimeSnapshot["probeByConnectionId"] extends Map ? T : never, - ][], -): HostRuntimeSnapshot["probeByConnectionId"] { - return new Map(entries); -} - describe("HostRuntimeController", () => { it("replaces the active relay client when re-pairing changes the daemon public key", async () => { const oldRelay: HostConnection = { @@ -339,11 +346,7 @@ describe("HostRuntimeController", () => { }, }); - await ( - controller as unknown as { - switchToConnection: (input: { connectionId: string }) => Promise; - } - ).switchToConnection({ connectionId: oldRelay.id }); + await controller.activateConnection({ connectionId: oldRelay.id }); expect(controller.getSnapshot().client).toBe(createdClients[0]?.client); await controller.updateHost( @@ -354,7 +357,7 @@ describe("HostRuntimeController", () => { ); expect(createdClients.map((entry) => entry.connection)).toEqual([oldRelay, newRelay]); - expect(createdClients[0]?.client.closeCalls).toBe(1); + expect(createdClients[0]?.client.isDisposed()).toBe(true); expect(controller.getSnapshot().client).toBe(createdClients[1]?.client); }); @@ -386,11 +389,7 @@ describe("HostRuntimeController", () => { // Intentionally do not emit a connected state; stay in idle. }; - await ( - controller as unknown as { - switchToConnection: (input: { connectionId: string }) => Promise; - } - ).switchToConnection({ connectionId: "direct:lan:6767" }); + await controller.activateConnection({ connectionId: "direct:lan:6767" }); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); expect(controller.getSnapshot().connectionStatus).toBe("connecting"); @@ -423,11 +422,7 @@ describe("HostRuntimeController", () => { }, }); - await ( - controller as unknown as { - switchToConnection: (input: { connectionId: string }) => Promise; - } - ).switchToConnection({ connectionId: "direct:lan:6767" }); + await controller.activateConnection({ connectionId: "direct:lan:6767" }); expect(seenClientIds).toEqual(["cid_runtime_stable"]); expect(controller.getSnapshot().connectionStatus).toBe("online"); @@ -453,7 +448,7 @@ describe("HostRuntimeController", () => { expect(clients).toHaveLength(2); expect(snapshot.client).toBe(clients[0] as unknown as DaemonClient); expect(clients[0]?.connectCalls).toBe(1); - expect(clients[1]?.closeCalls).toBe(1); + expect(clients[1]?.isDisposed()).toBe(true); }); it("activates the first successful probe without waiting for slower probes", async () => { @@ -504,7 +499,8 @@ describe("HostRuntimeController", () => { await probeCycle; }); - it("probes the active online connection through the existing client", async () => { + it("ranks the live connection by its heartbeat RTT without pinging it again", async () => { + useHostRuntimeClock(); const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); const probeAttempts: string[] = []; const latencies: Record = { @@ -541,17 +537,20 @@ describe("HostRuntimeController", () => { probeAttempts.length = 0; const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient; + activeClient.heartbeatReportsRtt(42); + activeClient.clearLatencyMeasurements(); activeClient.ping = async () => ({ rttMs: 9 }); - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(10_000); await controller.runProbeCycleNow(); - expect(probeAttempts).toEqual(["relay:relay.paseo.sh:443"]); + expect(probeAttempts).toEqual([]); 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, + latencyMs: 42, }); + expect(activeClient.latencyMeasurements()).toEqual([]); }); it("rejects probes that resolve to a different server id", async () => { @@ -589,10 +588,11 @@ describe("HostRuntimeController", () => { status: "unavailable", latencyMs: null, }); - expect(mismatchedClient.closeCalls).toBe(1); + expect(mismatchedClient.isDisposed()).toBe(true); }); - it("fails over when the active client ping fails", async () => { + it("keeps the live connection when one probe cycle looks slow", async () => { + useHostRuntimeClock(); const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); const clients: FakeDaemonClient[] = []; const latencies: Record = { @@ -609,60 +609,58 @@ 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"); + const activeClient = initialClient as unknown as FakeDaemonClient; + activeClient.heartbeatReportsRtt(200); + activeClient.latencyMeasurementsFailWith("active measurement failed"); latencies["relay:relay.paseo.sh:443"] = 42; - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); const snapshot = controller.getSnapshot(); - expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443"); + expect(snapshot.activeConnectionId).toBe("direct:lan:6767"); expect(snapshot.connectionStatus).toBe("online"); - expect(snapshot.client).not.toBe(initialClient); - expect((initialClient as unknown as FakeDaemonClient | null)?.closeCalls).toBe(1); + expect(snapshot.client).toBe(initialClient); + expect(activeClient.isDisposed()).toBe(false); }); - it("uses liveness probes instead of session RPC timeouts for active connection health", async () => { - const relay: HostConnection = { - id: "relay:relay.paseo.sh:443", - type: "relay", - relayEndpoint: "relay.paseo.sh:443", - useTls: true, - daemonPublicKeyB64: "pk_test", + it("does not mark the live connection unavailable before its first heartbeat resolves", async () => { + useHostRuntimeClock(); + const direct: HostConnection = { + id: "direct:lan:6767", + type: "directTcp", + endpoint: "lan:6767", }; const host = makeHost({ - connections: [relay], - preferredConnectionId: relay.id, + connections: [direct], + preferredConnectionId: direct.id, }); - const clients: FakeDaemonClient[] = []; + const activeClient = new FakeDaemonClient(); + activeClient.setConnectionState({ status: "connected" }); + activeClient.latencyMeasurementsFailWith("heartbeat has not resolved"); const controller = new HostRuntimeController({ host, - deps: makeDeps({ [relay.id]: 12 }, clients), + deps: makeDeps({ [direct.id]: 12 }, []), }); - await controller.start({ autoProbe: false }); - expect(controller.getSnapshot().connectionStatus).toBe("online"); - - const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient; - activeClient.ping = async () => { - throw new Error("Timeout waiting for message (5000ms)"); - }; - activeClient.checkLiveness = async () => ({ rttMs: 11 }); - clearProbeBackoff(controller); - - await controller.runProbeCycleNow(); + await controller.start({ + autoProbe: false, + initialConnection: { + connectionId: direct.id, + existingClient: activeClient as unknown as DaemonClient, + }, + }); const snapshot = controller.getSnapshot(); - expect(snapshot.probeByConnectionId.get(relay.id)).toEqual({ - status: "available", - latencyMs: 11, - }); + expect(snapshot.activeConnectionId).toBe(direct.id); expect(snapshot.connectionStatus).toBe("online"); + expect(snapshot.probeByConnectionId.get(direct.id)).toEqual({ + status: "pending", + latencyMs: null, + }); }); it("backs off inactive connection probes while a host is online", async () => { + useHostRuntimeClock(); const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); const clients: FakeDaemonClient[] = []; const latencies: Record = { @@ -676,6 +674,7 @@ describe("HostRuntimeController", () => { await controller.start({ autoProbe: false }); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient; const initialClientCount = clients.length; const initialRelayProbe = controller .getSnapshot() @@ -683,14 +682,8 @@ describe("HostRuntimeController", () => { latencies["direct:lan:6767"] = 12; latencies["relay:relay.paseo.sh:443"] = 25; - const lastProbedAt = ( - controller as unknown as { - connectionLastProbedAt: Map; - } - ).connectionLastProbedAt; - const now = performance.now(); - lastProbedAt.set("direct:lan:6767", now - 60_000); - lastProbedAt.set("relay:relay.paseo.sh:443", now - 60_000); + activeClient.heartbeatReportsRtt(12); + await vi.advanceTimersByTimeAsync(60_000); await controller.runProbeCycleNow(); @@ -704,6 +697,7 @@ describe("HostRuntimeController", () => { }); it("switches only after the faster alternative wins consecutive probes", async () => { + useHostRuntimeClock(); const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); const clients: FakeDaemonClient[] = []; const latencies: Record = { @@ -717,20 +711,22 @@ describe("HostRuntimeController", () => { await controller.start({ autoProbe: false }); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient; latencies["direct:lan:6767"] = 95; latencies["relay:relay.paseo.sh:443"] = 30; - clearProbeBackoff(controller); + activeClient.heartbeatReportsRtt(95); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); let switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443"; for (let index = 0; index < 6 && !switched; index += 1) { - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443"; } @@ -739,6 +735,7 @@ describe("HostRuntimeController", () => { }); it("does not switch on a transient latency spike", async () => { + useHostRuntimeClock(); const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); const clients: FakeDaemonClient[] = []; const latencies: Record = { @@ -752,32 +749,36 @@ describe("HostRuntimeController", () => { await controller.start({ autoProbe: false }); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + const activeClient = controller.getSnapshot().client as unknown as FakeDaemonClient; latencies["direct:lan:6767"] = 100; latencies["relay:relay.paseo.sh:443"] = 20; - clearProbeBackoff(controller); + activeClient.heartbeatReportsRtt(100); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); latencies["direct:lan:6767"] = 20; latencies["relay:relay.paseo.sh:443"] = 90; - clearProbeBackoff(controller); + activeClient.heartbeatReportsRtt(20); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); latencies["direct:lan:6767"] = 100; latencies["relay:relay.paseo.sh:443"] = 20; - clearProbeBackoff(controller); + activeClient.heartbeatReportsRtt(100); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); let switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443"; for (let index = 0; index < 6 && !switched; index += 1) { - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(120_000); await controller.runProbeCycleNow(); switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443"; } @@ -1041,11 +1042,7 @@ describe("HostRuntimeController", () => { } }; - const switchDirect = ( - controller as unknown as { - switchToConnection: (input: { connectionId: string }) => Promise; - } - ).switchToConnection({ connectionId: "direct:lan:6767" }); + const switchDirect = controller.activateConnection({ connectionId: "direct:lan:6767" }); await waitUntil(() => { const snapshot = controller.getSnapshot(); return ( @@ -1055,11 +1052,9 @@ describe("HostRuntimeController", () => { ); }); - const switchRelay = ( - controller as unknown as { - switchToConnection: (input: { connectionId: string }) => Promise; - } - ).switchToConnection({ connectionId: "relay:relay.paseo.sh:443" }); + const switchRelay = controller.activateConnection({ + connectionId: "relay:relay.paseo.sh:443", + }); await waitUntil(() => { const snapshot = controller.getSnapshot(); return ( @@ -1076,7 +1071,7 @@ describe("HostRuntimeController", () => { expect(snapshot.connectionStatus).toBe("online"); expect(snapshot.lastError).toBeNull(); expect(createdClients).toHaveLength(2); - expect(createdClients[0]?.closeCalls).toBe(1); + expect(createdClients[0]?.isDisposed()).toBe(true); }); it("coalesces overlapping probe cycles instead of invalidating the in-flight result", async () => { @@ -1118,7 +1113,6 @@ describe("HostRuntimeController", () => { }); const first = controller.runProbeCycleNow(); - clearProbeBackoff(controller); const second = controller.runProbeCycleNow(); expect(probeCalls).toBe(1); @@ -1132,6 +1126,7 @@ describe("HostRuntimeController", () => { }); it("keeps active client generation stable during background probe cycles", async () => { + useHostRuntimeClock(); const host = makeHost({ connections: [ { @@ -1167,140 +1162,12 @@ describe("HostRuntimeController", () => { const activeClientBeforeProbes = controller.getSnapshot().client; const generationBeforeProbes = controller.getSnapshot().clientGeneration; - clearProbeBackoff(controller); + await vi.advanceTimersByTimeAsync(10_000); await controller.runProbeCycleNow(); expect(controller.getSnapshot().client).toBe(activeClientBeforeProbes); expect(controller.getSnapshot().clientGeneration).toBe(generationBeforeProbes); expect(createdClients).toHaveLength(0); }); - - it("does not notify or replace the snapshot for equal probe maps", () => { - const controller = new HostRuntimeController({ host: makeHost() }); - const firstProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - const equalProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - - updateControllerSnapshot(controller, { probeByConnectionId: firstProbeMap }); - const snapshotAfterFirstProbe = controller.getSnapshot(); - let notifyCount = 0; - const unsubscribe = controller.subscribe(() => { - notifyCount += 1; - }); - - updateControllerSnapshot(controller, { probeByConnectionId: equalProbeMap }); - - expect(notifyCount).toBe(0); - expect(controller.getSnapshot()).toBe(snapshotAfterFirstProbe); - expect(controller.getSnapshot().probeByConnectionId).toBe(firstProbeMap); - unsubscribe(); - }); - - it("does not notify or replace the snapshot when connection status is already equal", () => { - const controller = new HostRuntimeController({ host: makeHost() }); - - updateControllerSnapshot(controller, { connectionStatus: "online" }); - const snapshotAfterOnline = controller.getSnapshot(); - let notifyCount = 0; - const unsubscribe = controller.subscribe(() => { - notifyCount += 1; - }); - - updateControllerSnapshot(controller, { connectionStatus: "online" }); - - expect(notifyCount).toBe(0); - expect(controller.getSnapshot()).toBe(snapshotAfterOnline); - unsubscribe(); - }); - - it("does not notify or replace the snapshot when every patched field is equal", () => { - const controller = new HostRuntimeController({ host: makeHost() }); - const firstProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - const equalProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - - updateControllerSnapshot(controller, { - connectionStatus: "online", - probeByConnectionId: firstProbeMap, - }); - const snapshotAfterSetup = controller.getSnapshot(); - let notifyCount = 0; - const unsubscribe = controller.subscribe(() => { - notifyCount += 1; - }); - - updateControllerSnapshot(controller, { - connectionStatus: "online", - probeByConnectionId: equalProbeMap, - }); - - expect(notifyCount).toBe(0); - expect(controller.getSnapshot()).toBe(snapshotAfterSetup); - expect(controller.getSnapshot().probeByConnectionId).toBe(firstProbeMap); - unsubscribe(); - }); - - it("notifies once for a changed field while preserving equal field identity", () => { - const controller = new HostRuntimeController({ host: makeHost() }); - const firstProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - const equalProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - - updateControllerSnapshot(controller, { - connectionStatus: "online", - probeByConnectionId: firstProbeMap, - }); - const snapshotBeforeChange = controller.getSnapshot(); - let notifyCount = 0; - const unsubscribe = controller.subscribe(() => { - notifyCount += 1; - }); - - updateControllerSnapshot(controller, { - connectionStatus: "offline", - probeByConnectionId: equalProbeMap, - }); - - expect(notifyCount).toBe(1); - expect(controller.getSnapshot()).not.toBe(snapshotBeforeChange); - expect(controller.getSnapshot().connectionStatus).toBe("offline"); - expect(controller.getSnapshot().probeByConnectionId).toBe(firstProbeMap); - unsubscribe(); - }); - - it("notifies once and replaces the probe map when probe contents change", () => { - const controller = new HostRuntimeController({ host: makeHost() }); - const firstProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ]); - const changedProbeMap = makeProbeMap([ - ["direct:lan:6767", { status: "available", latencyMs: 12 }], - ["relay:relay.paseo.sh:443", { status: "unavailable", latencyMs: null }], - ]); - - updateControllerSnapshot(controller, { probeByConnectionId: firstProbeMap }); - const snapshotBeforeChange = controller.getSnapshot(); - let notifyCount = 0; - const unsubscribe = controller.subscribe(() => { - notifyCount += 1; - }); - - updateControllerSnapshot(controller, { probeByConnectionId: changedProbeMap }); - - expect(notifyCount).toBe(1); - expect(controller.getSnapshot()).not.toBe(snapshotBeforeChange); - expect(controller.getSnapshot().probeByConnectionId).toBe(changedProbeMap); - expect(controller.getSnapshot().probeByConnectionId).not.toBe(firstProbeMap); - unsubscribe(); - }); }); describe("HostRuntimeStore", () => { @@ -1770,7 +1637,7 @@ describe("HostRuntimeStore", () => { expect(result.serverId).toBe("srv_real_direct"); expect(result.hostname).toBe("mbp"); expect(seenProbeHosts).toEqual([""]); - expect(probeClient.closeCalls).toBe(0); + expect(probeClient.isDisposed()).toBe(false); expect(store.getHosts()).toMatchObject([ { serverId: "srv_real_direct", diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index c84bf13f2..62da522ee 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -674,7 +674,7 @@ export class HostRuntimeController { async activateConnection(input: { connectionId: string; - existingClient: DaemonClient; + existingClient?: DaemonClient; }): Promise { await this.switchToConnection(input); } @@ -913,7 +913,22 @@ export class HostRuntimeController { const activated = await maybeActivateFirstAvailable(connection.id, connectedClient); shouldCloseClient = shouldCloseClient && !activated; - const { rttMs } = await connectedClient.checkLiveness({ timeoutMs: 5000 }); + if (activeClient) { + const rttMs = activeClient.getLastLivenessRttMs(); + if (!this.isCurrentProbeRequest(requestVersion)) { + return; + } + if (rttMs !== null) { + probeByConnectionId.set(connection.id, { + status: "available", + latencyMs: rttMs, + }); + publishProbeState(); + } + return; + } + + const rttMs = await connectedClient.measureLatency({ timeoutMs: 5000 }); if (!this.isCurrentProbeRequest(requestVersion)) { return; } diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index e0c07fbc5..8bae233f3 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -1,6 +1,6 @@ import { afterEach, expect, expectTypeOf, test, vi } from "vitest"; import { z } from "zod"; -import { DaemonClient, type DaemonTransport } from "./daemon-client"; +import { DaemonClient, type DaemonTransport, type Logger } from "./daemon-client"; import { decodeFileTransferFrame, encodeFileTransferFrame, @@ -42,7 +42,16 @@ function createMockTransport() { let serverInfoOrdinal = 1; const transport: DaemonTransport = { - send: (data) => sent.push(data), + send: (data) => { + sent.push(data); + if (typeof data !== "string") { + return; + } + const frame = JSON.parse(data) as { type?: string }; + if (frame.type === "ping") { + onMessage(JSON.stringify({ type: "pong" })); + } + }, close: () => {}, onMessage: (handler) => { onMessage = handler; @@ -126,8 +135,205 @@ const clients: DaemonClient[] = []; afterEach(async () => { await Promise.all(clients.map((client) => client.close())); clients.length = 0; + vi.useRealTimers(); }); +const noopLogger: Logger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: () => {}, +}; + +type PongMode = { kind: "answer"; delayMs: number } | { kind: "silent" }; + +class FakeDaemon { + private onMessage: (data: unknown) => void = () => {}; + private onOpen: () => void = () => {}; + private onClose: (event?: unknown) => void = () => {}; + private onError: (event?: unknown) => void = () => {}; + private pongMode: PongMode = { kind: "answer", delayMs: 0 }; + private withheldPongs = 0; + private pingsSentAt: number[] = []; + private closeEvents: Array<{ code?: number; reason?: string }> = []; + + readonly transport: DaemonTransport = { + send: (data) => { + if (typeof data !== "string") { + return; + } + const frame = JSON.parse(data) as { type?: string }; + if (frame.type !== "ping") { + return; + } + this.pingsSentAt.push(performance.now()); + if (this.withheldPongs > 0) { + this.withheldPongs -= 1; + return; + } + if (this.pongMode.kind === "silent") { + return; + } + if (this.pongMode.delayMs === 0) { + this.onMessage(JSON.stringify({ type: "pong" })); + return; + } + setTimeout(() => { + this.onMessage(JSON.stringify({ type: "pong" })); + }, this.pongMode.delayMs); + }, + close: (code?: number, reason?: string) => { + this.closeEvents.push({ code, reason }); + }, + onMessage: (handler) => { + this.onMessage = handler; + return () => {}; + }, + onOpen: (handler) => { + this.onOpen = handler; + return () => {}; + }, + onClose: (handler) => { + this.onClose = handler; + return () => {}; + }, + onError: (handler) => { + this.onError = handler; + return () => {}; + }, + }; + + openConnection(): void { + this.onOpen(); + this.onMessage( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "srv_heartbeat_test", + hostname: null, + version: null, + }, + }, + }), + ); + } + + daemonAnswersPingsAfter(delay: "fast" | `${number}s`): void { + this.pongMode = { + kind: "answer", + delayMs: delay === "fast" ? 0 : Number.parseFloat(delay) * 1000, + }; + } + + daemonGoesSilent(): void { + this.pongMode = { kind: "silent" }; + } + + daemonWithholdsNextPongThenAnswersFast(): void { + this.withheldPongs += 1; + this.daemonAnswersPingsAfter("fast"); + } + + daemonClosesWith(reason: string): void { + this.onClose({ reason }); + } + + triggerError(event?: unknown): void { + this.onError(event); + } + + pingTimestamps(): string[] { + return this.pingsSentAt.map((timestamp) => `${timestamp / 1000}s`); + } + + teardownCount(): number { + return this.closeEvents.filter((event) => event.reason === "Liveness check timed out").length; + } + + closesFromClient(): Array<{ code?: number; reason?: string }> { + return this.closeEvents; + } +} + +class DaemonClientSession { + private readonly daemon = new FakeDaemon(); + private readonly client: DaemonClient; + + constructor() { + this.client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_heartbeat_test", + logger: noopLogger, + reconnect: { enabled: false }, + transportFactory: () => this.daemon.transport, + }); + clients.push(this.client); + } + + async connect(): Promise { + const connection = this.client.connect(); + this.daemon.openConnection(); + await connection; + } + + async advance(ms: number): Promise { + await vi.advanceTimersByTimeAsync(ms); + } + + daemonAnswersPingsAfter(delay: "fast" | `${number}s`): void { + this.daemon.daemonAnswersPingsAfter(delay); + } + + daemonGoesSilent(): void { + this.daemon.daemonGoesSilent(); + } + + daemonWithholdsNextPongThenAnswersFast(): void { + this.daemon.daemonWithholdsNextPongThenAnswersFast(); + } + + daemonClosesWith(reason: string): void { + this.daemon.daemonClosesWith(reason); + } + + pingTimestamps(): string[] { + return this.daemon.pingTimestamps(); + } + + state(): ReturnType { + return this.client.getConnectionState(); + } + + lastError(): string | null { + return this.client.lastError; + } + + teardownCount(): number { + return this.daemon.teardownCount(); + } + + closesFromClient(): Array<{ code?: number; reason?: string }> { + return this.daemon.closesFromClient(); + } + + lastLivenessRttMs(): number | null { + return this.client.getLastLivenessRttMs(); + } + + measureLatency(input: { timeoutMs: number }): Promise { + return this.client.measureLatency(input); + } +} + +function useHeartbeatClock(): void { + vi.useFakeTimers({ + toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval", "performance"], + }); +} + test("dedupes in-flight checkout status requests per agentId", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -316,59 +522,205 @@ test("keeps the transport connected when a session RPC ping times out", async () expect(client.getConnectionState().status).toBe("connected"); }); -test("reconnects after repeated top-level liveness checks time out", async () => { - const logger = createMockLogger(); - const mock = createMockTransport(); +test("stays online through ten minutes of pongs that arrive five seconds late", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("5.5s"); - const client = new DaemonClient({ - url: "ws://test", - clientId: "clsk_unit_test", - logger, - reconnect: { enabled: false }, - transportFactory: () => mock.transport, - }); - clients.push(client); + await session.connect(); + await session.advance(10 * 60 * 1000); - const connectPromise = client.connect(); - mock.triggerOpen(); - await connectPromise; - expect(client.getConnectionState().status).toBe("connected"); - - await expect(client.checkLiveness({ timeoutMs: 1 })).rejects.toThrow("Liveness check timed out"); - expect(client.getConnectionState().status).toBe("connected"); - - await expect(client.checkLiveness({ timeoutMs: 1 })).rejects.toThrow("Liveness check timed out"); - expect(client.getConnectionState()).toEqual({ - status: "disconnected", - reason: "Liveness check timed out (1ms)", - }); + expect(session.state()).toEqual({ status: "connected" }); + expect(session.teardownCount()).toBe(0); + expect(session.lastError()).toBeNull(); + expect(session.pingTimestamps().length).toBeGreaterThan(0); + expect(session.lastLivenessRttMs()).toBe(5500); }); -test("resets liveness failures after a top-level pong", async () => { - const logger = createMockLogger(); - const mock = createMockTransport(); +test("tears down and reports a liveness timeout after the daemon goes silent for two cycles", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonGoesSilent(); - const client = new DaemonClient({ - url: "ws://test", - clientId: "clsk_unit_test", - logger, - reconnect: { enabled: false }, - transportFactory: () => mock.transport, + await session.connect(); + await session.advance(51_000); + + expect(session.state()).toEqual({ + status: "disconnected", + reason: "Liveness check timed out (15000ms)", }); - clients.push(client); + expect(session.teardownCount()).toBe(1); +}); - const connectPromise = client.connect(); - mock.triggerOpen(); - await connectPromise; +test("survives a single missed pong when answers resume", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonWithholdsNextPongThenAnswersFast(); - await expect(client.checkLiveness({ timeoutMs: 1 })).rejects.toThrow("Liveness check timed out"); + await session.connect(); + await session.advance(40_000); - const healthyProbe = client.checkLiveness({ timeoutMs: 100 }); - mock.triggerMessage(JSON.stringify({ type: "pong" })); - await expect(healthyProbe).resolves.toEqual({ rttMs: expect.any(Number) }); + expect(session.state()).toEqual({ status: "connected" }); + expect(session.teardownCount()).toBe(0); +}); - await expect(client.checkLiveness({ timeoutMs: 1 })).rejects.toThrow("Liveness check timed out"); - expect(client.getConnectionState().status).toBe("connected"); +test("keeps the connection proven online from heartbeats alone when no other traffic flows", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("fast"); + + await session.connect(); + await session.advance(35_000); + + expect(session.state()).toEqual({ status: "connected" }); + expect(session.pingTimestamps()).toEqual(["10s", "20s", "30s"]); +}); + +test("starts pinging one interval after connecting and holds the cadence", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("fast"); + + await session.connect(); + await session.advance(9_999); + expect(session.pingTimestamps()).toEqual([]); + + await session.advance(20_001); + expect(session.pingTimestamps()).toEqual(["10s", "20s", "30s"]); +}); + +test("stops pinging once the connection is gone", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("fast"); + + await session.connect(); + await session.advance(10_000); + session.daemonClosesWith("daemon shutting down"); + await session.advance(30_000); + + expect(session.pingTimestamps()).toEqual(["10s"]); +}); + +test("sends only one ping while a slow pong is still outstanding", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("15.1s"); + + await session.connect(); + await session.advance(20_000); + + expect(session.pingTimestamps()).toEqual(["10s"]); +}); + +test("reports the round-trip time of the last successful heartbeat", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonAnswersPingsAfter("2.3s"); + + await session.connect(); + await session.advance(12_300); + + expect(session.lastLivenessRttMs()).toBe(2300); +}); + +test("treats a pong just under the timeout as alive and just over as a miss", async () => { + useHeartbeatClock(); + const alive = new DaemonClientSession(); + alive.daemonAnswersPingsAfter("14.9s"); + + await alive.connect(); + await alive.advance(24_900); + + expect(alive.state()).toEqual({ status: "connected" }); + + const missed = new DaemonClientSession(); + missed.daemonAnswersPingsAfter("15.1s"); + + await missed.connect(); + await missed.advance(25_000); + + expect(missed.state()).toEqual({ status: "connected" }); + expect(missed.lastLivenessRttMs()).toBeNull(); +}); + +test("goes red with the daemon's reason when the connection is closed", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + + await session.connect(); + session.daemonClosesWith("Control unresponsive"); + + expect(session.state()).toEqual({ + status: "disconnected", + reason: "Control unresponsive", + }); + expect(session.lastError()).toBe("Control unresponsive"); + expect(session.closesFromClient()).toEqual([]); +}); + +test("two candidate latency timeouts do not tear down the live connection", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonGoesSilent(); + + await session.connect(); + const firstMeasurement = session.measureLatency({ timeoutMs: 1000 }); + const firstMeasurementError = firstMeasurement.then( + () => null, + (error) => error, + ); + await session.advance(1000); + + await expect(firstMeasurementError).resolves.toEqual( + expect.objectContaining({ + message: "Latency measurement timed out (1000ms)", + }), + ); + + const secondMeasurement = session.measureLatency({ timeoutMs: 1000 }); + const secondMeasurementError = secondMeasurement.then( + () => null, + (error) => error, + ); + await session.advance(1000); + + await expect(secondMeasurementError).resolves.toEqual( + expect.objectContaining({ + message: "Latency measurement timed out (1000ms)", + }), + ); + expect(session.state()).toEqual({ status: "connected" }); + expect(session.teardownCount()).toBe(0); + expect(session.pingTimestamps()).toEqual(["0s", "1s"]); +}); + +test("a candidate measurement that times out under a heartbeat tick does not count toward teardown", async () => { + useHeartbeatClock(); + const session = new DaemonClientSession(); + session.daemonGoesSilent(); + + await session.connect(); + + // A candidate measurement is still in flight when the +10s heartbeat tick lands, + // so the heartbeat shares the in-flight ping. Let the measurement time out. + await session.advance(9_000); + const measurement = session.measureLatency({ timeoutMs: 5_000 }); + const measurementError = measurement.then( + () => null, + (error) => error, + ); + await session.advance(5_500); + await expect(measurementError).resolves.toEqual( + expect.objectContaining({ message: "Latency measurement timed out (5000ms)" }), + ); + + // The measurement timeout must not have been recorded as a liveness failure: a + // single genuine heartbeat miss after it must still leave the connection up. + await session.advance(25_000); + + expect(session.state()).toEqual({ status: "connected" }); + expect(session.teardownCount()).toBe(0); }); test("listDirectory sends a list file explorer request and returns directory entries", async () => { @@ -1389,7 +1741,7 @@ test("restartServer remains restart-only and sends restart_server_request", asyn }); test("transitions out of connecting when connect timeout elapses", async () => { - vi.useFakeTimers(); + useHeartbeatClock(); try { const logger = createMockLogger(); const mock = createMockTransport(); @@ -1426,7 +1778,7 @@ test("transitions out of connecting when connect timeout elapses", async () => { }); test("reconnects after relay close with replaced-by-new-connection reason", async () => { - vi.useFakeTimers(); + useHeartbeatClock(); try { const logger = createMockLogger(); const first = createMockTransport(); @@ -2658,7 +3010,7 @@ test("imports an agent by provider handle id", async () => { }); test("uses server-provided dictation finish timeout budget", async () => { - vi.useFakeTimers(); + useHeartbeatClock(); const logger = createMockLogger(); const mock = createMockTransport(); @@ -2748,7 +3100,7 @@ test("resolves dictation finish when final arrives after finish accepted", async }); test("cancels waiters when send fails (no leaked timeouts)", async () => { - vi.useFakeTimers(); + useHeartbeatClock(); const logger = createMockLogger(); const mock = createMockTransport(); let sendCount = 0; @@ -2783,7 +3135,7 @@ test("cancels waiters when send fails (no leaked timeouts)", async () => { const internal = client as unknown as { waiters: Set }; expect(internal.waiters.size).toBe(0); - vi.runOnlyPendingTimers(); + await vi.advanceTimersByTimeAsync(0); vi.useRealTimers(); }); @@ -3689,7 +4041,7 @@ test("sends close_items_request and resolves close_items_response", async () => }); test("waitForFinish with timeout=0 omits timeoutMs and has no client deadline", async () => { - vi.useFakeTimers(); + useHeartbeatClock(); try { const logger = createMockLogger(); const mock = createMockTransport(); @@ -3715,14 +4067,20 @@ test("waitForFinish with timeout=0 omits timeoutMs and has no client deadline", expect(request.agentId).toBe("agent-wait-zero-timeout"); expect(request).not.toHaveProperty("timeoutMs"); - const settled = vi.fn(); + let settled: "pending" | "resolved" | "rejected" = "pending"; void waitPromise.then( - () => settled("resolved"), - () => settled("rejected"), + () => { + settled = "resolved"; + return null; + }, + () => { + settled = "rejected"; + return null; + }, ); await vi.advanceTimersByTimeAsync(5 * 60 * 1000); - expect(settled).not.toHaveBeenCalled(); + expect(settled).toBe("pending"); mock.triggerMessage( wrapSessionMessage({ diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index dfd17220e..e8507b6c7 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -737,10 +737,26 @@ class DaemonRpcError extends Error { } } +class PingTimeoutError extends Error { + constructor(readonly timeoutMs: number) { + super(`Ping timed out (${timeoutMs}ms)`); + this.name = "PingTimeoutError"; + } +} + +function toTimeoutError(error: unknown, label: string, timeoutMs: number): Error { + if (error instanceof PingTimeoutError) { + return new Error(`${label} timed out (${timeoutMs}ms)`); + } + return error instanceof Error ? error : new Error(String(error)); +} + const DEFAULT_RECONNECT_BASE_DELAY_MS = 1500; const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000; const DEFAULT_CONNECT_TIMEOUT_MS = 15000; const DEFAULT_LIVENESS_TIMEOUT_MS = 5000; +const LIVENESS_HEARTBEAT_INTERVAL_MS = 10_000; +const LIVENESS_HEARTBEAT_TIMEOUT_MS = 15_000; const LIVENESS_FAILURE_RECONNECT_THRESHOLD = 2; /** Default timeout for waiting for connection before sending queued messages */ @@ -848,12 +864,16 @@ interface PendingSend { timeoutHandle: ReturnType; } -interface LivenessProbe { - promise: Promise<{ rttMs: number }>; - resolve: (value: { rttMs: number }) => void; +interface PingProbe { + promise: Promise; + resolve: (value: number) => void; reject: (error: Error) => void; timeoutHandle: ReturnType; startedAt: number; + // Whether a timeout on this ping should be recorded as a liveness failure. Only the + // heartbeat sets this; a latency measurement never drives teardown, even when a + // heartbeat tick shares (dedupes onto) an in-flight measurement ping. + drivesLivenessFailure: boolean; } export class DaemonClient { @@ -899,7 +919,9 @@ export class DaemonClient { private lastServerInfoMessage: ServerInfoStatusPayload | null = null; private runtimeMetricsInterval: ReturnType | null = null; private runtimeMetrics: DaemonClientRuntimeMetrics | null = null; - private livenessProbe: LivenessProbe | null = null; + private pingProbe: PingProbe | null = null; + private livenessHeartbeatTimer: ReturnType | null = null; + private lastLivenessRttMs: number | null = null; private consecutiveLivenessFailures = 0; constructor(private config: DaemonClientConfig) { @@ -1162,7 +1184,7 @@ export class DaemonClient { this.disposeTransport(1000, "Client closed"); this.clearWaiters(new Error("Daemon client closed")); this.rejectPendingSendQueue(new Error("Daemon client closed")); - this.rejectLivenessProbe(new Error("Daemon client closed")); + this.rejectPingProbe(new Error("Daemon client closed")); this.terminalStreams.clearSlots(); this.lastServerInfoMessage = null; if (this.runtimeMetricsInterval) { @@ -1217,6 +1239,10 @@ export class DaemonClient { return this.lastErrorValue; } + getLastLivenessRttMs(): number | null { + return this.lastLivenessRttMs; + } + // ============================================================================ // Message Subscription // ============================================================================ @@ -1627,54 +1653,108 @@ export class DaemonClient { }; } - checkLiveness(params?: { timeoutMs?: number }): Promise<{ rttMs: number }> { + measureLatency(params?: { timeoutMs?: number }): Promise { + const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS); + return this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: false }).catch((error) => { + throw toTimeoutError(error, "Latency measurement", timeoutMs); + }); + } + + private async livenessPing(params?: { timeoutMs?: number }): Promise { + const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS); + try { + const rttMs = await this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: true }); + this.lastLivenessRttMs = rttMs; + return rttMs; + } catch (error) { + throw toTimeoutError(error, "Liveness check", timeoutMs); + } + } + + private sendPingAwaitRtt(params: { + timeoutMs: number; + drivesLivenessFailure: boolean; + }): Promise { if (this.connectionState.status !== "connected" || !this.transport) { return Promise.reject( new Error(`Transport not connected (status: ${this.connectionState.status})`), ); } - if (this.livenessProbe) { - return this.livenessProbe.promise; + if (this.pingProbe) { + return this.pingProbe.promise; } const startedAt = perfNow(); - const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS); - let resolveProbe: ((value: { rttMs: number }) => void) | null = null; + const timeoutMs = params.timeoutMs; + let resolveProbe: ((value: number) => void) | null = null; let rejectProbe: ((error: Error) => void) | null = null; - const promise = new Promise<{ rttMs: number }>((resolve, reject) => { + const promise = new Promise((resolve, reject) => { resolveProbe = resolve; rejectProbe = reject; }); - const probe: LivenessProbe = { + const probe: PingProbe = { promise, resolve: (value) => resolveProbe?.(value), reject: (error) => rejectProbe?.(error), timeoutHandle: setTimeout(() => { - if (this.livenessProbe !== probe) { + if (this.pingProbe !== probe) { return; } - this.livenessProbe = null; - const error = new Error(`Liveness check timed out (${timeoutMs}ms)`); + this.pingProbe = null; + const error = new PingTimeoutError(timeoutMs); probe.reject(error); - this.recordLivenessFailure(error); + if (probe.drivesLivenessFailure) { + this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs)); + } }, timeoutMs), startedAt, + drivesLivenessFailure: params.drivesLivenessFailure, }; - this.livenessProbe = probe; + this.pingProbe = probe; try { this.transport.send(JSON.stringify({ type: "ping" })); } catch (error) { - this.clearLivenessProbe(); - const err = error instanceof Error ? error : new Error(String(error)); - this.recordLivenessFailure(err); - return Promise.reject(err); + this.clearPingProbe(); + const sendError = error instanceof Error ? error : new Error(String(error)); + if (probe.drivesLivenessFailure) { + this.recordLivenessFailure(sendError); + } + return Promise.reject(sendError); } return promise; } + private startLivenessHeartbeat(): void { + this.stopLivenessHeartbeat(); + this.lastLivenessRttMs = null; + this.scheduleNextLivenessHeartbeat(); + } + + private stopLivenessHeartbeat(): void { + if (!this.livenessHeartbeatTimer) { + return; + } + clearTimeout(this.livenessHeartbeatTimer); + this.livenessHeartbeatTimer = null; + } + + private scheduleNextLivenessHeartbeat(): void { + if (this.connectionState.status !== "connected" || this.livenessHeartbeatTimer) { + return; + } + this.livenessHeartbeatTimer = setTimeout(() => { + this.livenessHeartbeatTimer = null; + this.livenessPing({ timeoutMs: LIVENESS_HEARTBEAT_TIMEOUT_MS }) + .catch(() => {}) + .finally(() => { + this.scheduleNextLivenessHeartbeat(); + }); + }, LIVENESS_HEARTBEAT_INTERVAL_MS); + } + // ============================================================================ // Agent RPCs (requestId-correlated) // ============================================================================ @@ -4463,6 +4543,7 @@ export class DaemonClient { } private disposeTransport(code = 1001, reason = "Reconnecting"): void { + this.stopLivenessHeartbeat(); this.cleanupTransport(); if (this.transport) { try { @@ -4556,7 +4637,7 @@ export class DaemonClient { this.consecutiveLivenessFailures = 0; if (parsed.data.type === "pong") { - this.resolveLivenessProbe(); + this.resolvePingProbe(); this.runtimeMetrics?.recordMessage("pong", bytes, perfNow() - startMs); return; } @@ -4572,6 +4653,7 @@ export class DaemonClient { private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean { const fileFrame = decodeFileTransferFrame(rawBytes); if (fileFrame) { + this.consecutiveLivenessFailures = 0; this.handleFileTransferFrame(fileFrame); this.runtimeMetrics?.recordBinaryFrame("other", rawBytes.byteLength, 0); return true; @@ -4581,6 +4663,7 @@ export class DaemonClient { if (!frame) { return false; } + this.consecutiveLivenessFailures = 0; const binaryStartMs = perfNow(); this.terminalStreams.handleFrame(frame); let frameKind: "output" | "snapshot" | "other" = "other"; @@ -4707,7 +4790,7 @@ export class DaemonClient { // and responses from the previous connection will never arrive. this.clearWaiters(new Error(reason ?? "Connection lost")); this.rejectPendingSendQueue(new Error(reason ?? "Connection lost")); - this.rejectLivenessProbe(new Error(reason ?? "Connection lost")); + this.rejectPingProbe(new Error(reason ?? "Connection lost")); this.terminalStreams.clearSlots(); this.lastServerInfoMessage = null; @@ -4756,31 +4839,31 @@ export class DaemonClient { }, delay); } - private resolveLivenessProbe(): void { - const probe = this.livenessProbe; + private resolvePingProbe(): void { + const probe = this.pingProbe; if (!probe) { return; } - this.livenessProbe = null; + this.pingProbe = null; clearTimeout(probe.timeoutHandle); - probe.resolve({ rttMs: perfNow() - probe.startedAt }); + probe.resolve(perfNow() - probe.startedAt); } - private clearLivenessProbe(): void { - const probe = this.livenessProbe; + private clearPingProbe(): void { + const probe = this.pingProbe; if (!probe) { return; } - this.livenessProbe = null; + this.pingProbe = null; clearTimeout(probe.timeoutHandle); } - private rejectLivenessProbe(error: Error): void { - const probe = this.livenessProbe; + private rejectPingProbe(error: Error): void { + const probe = this.pingProbe; if (!probe) { return; } - this.livenessProbe = null; + this.pingProbe = null; clearTimeout(probe.timeoutHandle); probe.reject(error); } @@ -4809,6 +4892,7 @@ export class DaemonClient { this.resetConnectTimeout(); this.reconnectAttempt = 0; this.updateConnectionState({ status: "connected" }, { event: "HELLO_SERVER_INFO" }); + this.startLivenessHeartbeat(); this.resubscribeCheckoutDiffSubscriptions(); this.resubscribeTerminalDirectorySubscriptions(); this.flushPendingSendQueue();