From 85f7c14ca30a06172cd903058fefa28cb38a345d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 27 Oct 2026 00:52:21 -0700 Subject: [PATCH 1/2] fix(client-runtime): resume keeps live sockets and replaces dead ones fast A resume judged the socket by the clock: a short resume waited out a 3 s probe on a socket iOS had killed, a long one replaced even a healthy socket and resynced, and a slow remote route lost a live socket after 3 s. Every resume now probes the socket. A probe unanswered after 500 ms asks the host for its public descriptor over a fresh request, retrying each second. If the host answers while the socket delivers no frame, the socket is replaced: at once on loopback and LAN, after 2.5 s elsewhere. Remote routes give a live socket 15 s on a short resume or network change. Co-authored-by: Omar <76748259+omar9848492@users.noreply.github.com> Co-authored-by: Carlos D. Escobar-Valbuena Co-Authored-By: Claude Opus 5.5 --- .../src/connection/app-state-wakeups.ts | 8 +- apps/mobile/src/connection/platform.ts | 2 +- apps/web/src/cloud/linkEnvironment.test.ts | 1 + apps/web/test/fakeEnvironment.ts | 2 + docs/internals/connection-runtime.md | 25 +- .../client-runtime/src/connection/driver.ts | 29 +- .../src/connection/registry.test.ts | 2 + .../src/connection/supervisor.test.ts | 352 +++++++++++++++++- .../src/connection/supervisor.ts | 157 +++++++- .../client-runtime/src/connection/wakeups.ts | 7 +- .../operations/commands.performance.test.ts | 1 + .../src/operations/commands.test.ts | 1 + .../client-runtime/src/rpc/client.test.ts | 1 + .../client-runtime/src/rpc/session.test.ts | 17 + packages/client-runtime/src/rpc/session.ts | 17 +- .../src/state/pullRequests.test.ts | 1 + .../client-runtime/src/state/server.test.ts | 1 + .../src/state/serverUsage.test.ts | 1 + .../src/state/shell-sync.test.ts | 1 + .../src/state/threads-atoms.test.ts | 1 + .../src/state/threads-sync.test.ts | 1 + packages/client-runtime/src/state/vcs.test.ts | 1 + .../src/state/vcsAction.test.ts | 1 + 23 files changed, 566 insertions(+), 64 deletions(-) diff --git a/apps/mobile/src/connection/app-state-wakeups.ts b/apps/mobile/src/connection/app-state-wakeups.ts index 185ff744958ed..d3f5b225a1671 100644 --- a/apps/mobile/src/connection/app-state-wakeups.ts +++ b/apps/mobile/src/connection/app-state-wakeups.ts @@ -1,10 +1,10 @@ import type { Wakeups } from "@t3tools/client-runtime/connection"; -export const MOBILE_BACKGROUND_RECONNECT_AFTER_MS = 10_000; +export const MOBILE_LONG_BACKGROUND_AFTER_MS = 10_000; export type MobileApplicationActiveWakeup = Extract< Wakeups.ConnectionWakeup, - "application-active-probe" | "application-active-reconnect" + "application-active-probe" | "application-active-long" >; export function mobileApplicationActiveWakeup( @@ -12,7 +12,7 @@ export function mobileApplicationActiveWakeup( activeAtMs: number, ): MobileApplicationActiveWakeup { return backgroundedAtMs !== null && - activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS - ? "application-active-reconnect" + activeAtMs - backgroundedAtMs >= MOBILE_LONG_BACKGROUND_AFTER_MS + ? "application-active-long" : "application-active-probe"; } diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index 52a6106436389..b99b41cbffb7b 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -118,7 +118,7 @@ const networkPathChanges = Stream.callback<"network-changed">((queue) => const layerWakeups = Wakeups.layer({ changes: Stream.mergeAll( [ - Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) => + Stream.callback<"application-active-probe" | "application-active-long">((queue) => Effect.acquireRelease( Effect.sync(() => { let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null; diff --git a/apps/web/src/cloud/linkEnvironment.test.ts b/apps/web/src/cloud/linkEnvironment.test.ts index 65cf9402e5fc4..68b2c92c05c1a 100644 --- a/apps/web/src/cloud/linkEnvironment.test.ts +++ b/apps/web/src/cloud/linkEnvironment.test.ts @@ -85,6 +85,7 @@ function layerRegistry(options?: { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; const target = new PrimaryConnectionTarget({ diff --git a/apps/web/test/fakeEnvironment.ts b/apps/web/test/fakeEnvironment.ts index 318a8cd0267b9..b9d6b43bf4ec4 100644 --- a/apps/web/test/fakeEnvironment.ts +++ b/apps/web/test/fakeEnvironment.ts @@ -377,6 +377,7 @@ const session: EnvironmentConnectionLease["session"] = { subscribe([{ version: 1 as const, type: "snapshot" as const, config: serverConfig }]), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; @@ -398,6 +399,7 @@ export const fakeDriver = { }), checkRoute: () => Effect.succeed("answered"), preflight: () => Effect.succeed(true), + checkHost: () => Effect.succeed("unchecked"), }; export type FakeDriver = typeof fakeDriver; diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 1f3df3ba0aff6..b614b6b515df3 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -16,14 +16,23 @@ client of a restarted server reconnects in the same second; with a short cap, a client that can never connect retries all day. Offline states and authentication failures wait for a wakeup instead of spending attempts on unchanged conditions. -Foregrounding, an explicit retry, and an offline report probe the established -session, and only a failed probe reconnects. Offline reports are often wrong, for -example for a loopback server. A long mobile background suspension is the one -exception: it replaces the session at once, because the OS can kill a socket -without reporting closure, and a probe would hold a dead socket in "Resuming" -until it times out. That fresh attempt runs even while the network reports -offline. Foregrounding also wakes a pending retry immediately and -leaves an ordinary in-flight attempt alone. +Foregrounding, a network change, an explicit retry, and an offline report probe +the established session, and only a failed probe reconnects. Offline reports are +often wrong, for example for a loopback server. The OS can kill a suspended +socket without a close event, so a probe still unanswered after 500 ms asks the +host for its public descriptor over a fresh request, again each second until it +answers. If the host answers while the socket delivers no frame, the socket is +dead and is replaced, even while the network reports offline: at once on +loopback and LAN, and 2.5 seconds after the check starts elsewhere, where one +lost packet can stall a live socket. Without that evidence the probe deadline +decides: 3 seconds on loopback and LAN routes and after a long mobile +suspension, when the data on screen is stale; 15 seconds elsewhere, because a +tailnet over DERP or a VPN can be slower than that on a live socket. A long +suspension's probe that times out also replaces the socket regardless of the +network report. The web and desktop foreground probe skips the fresh request, +because a busy host can answer it before a queued probe. Foregrounding also +wakes a pending retry immediately. A long suspension restarts an in-flight +attempt; a short one leaves it alone. The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes connections by environment. An involuntary disconnect retains the registration diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index f592ac7ea2f1b..7a08368096e58 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -61,6 +61,8 @@ export class ConnectionDriver extends Context.Service< entry: ConnectionCatalogEntry, route: ConnectionRoute, ) => Effect.Effect; + /** Asks the host behind a live session to answer a fresh request on the session's own origin. */ + readonly checkHost: (prepared: PreparedConnection) => Effect.Effect; } >()("@t3tools/client-runtime/connection/driver/ConnectionDriver") {} @@ -142,26 +144,37 @@ export const make = Effect.gen(function* () { const sessions = yield* RpcSession.RpcSessionFactory; const httpClient = yield* HttpClient.HttpClient; - const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => { - const httpBaseUrl = routeHttpBaseUrl(route); - if (httpBaseUrl === null) return Effect.succeed("unchecked"); - // The descriptor is public, so this sends no credential to whatever - // answers at a saved LAN address on a different network. - return fetchRemoteEnvironmentDescriptor({ + // The descriptor is public, so this sends no credential to whatever + // answers at a saved LAN address on a different network. + const checkDescriptor = (httpBaseUrl: string, environmentId: string) => + fetchRemoteEnvironmentDescriptor({ httpBaseUrl, timeoutMs: ROUTE_CHECK_TIMEOUT_MS, }).pipe( Effect.map((descriptor): RouteCheck => - descriptor.environmentId === entry.target.environmentId ? "answered" : "silent", + descriptor.environmentId === environmentId ? "answered" : "silent", ), Effect.orElseSucceed((): RouteCheck => "silent"), Effect.provideService(HttpClient.HttpClient, httpClient), + ); + + const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => { + const httpBaseUrl = routeHttpBaseUrl(route); + if (httpBaseUrl === null) return Effect.succeed("unchecked"); + return checkDescriptor(httpBaseUrl, entry.target.environmentId).pipe( Effect.withSpan("ConnectionDriver.checkRoute", { attributes: { "connection.target.kind": route.target._tag }, }), ); }; + const checkHost = (prepared: PreparedConnection) => + checkDescriptor(prepared.httpBaseUrl, prepared.environmentId).pipe( + Effect.withSpan("ConnectionDriver.checkHost", { + attributes: { "connection.target.kind": prepared.target._tag }, + }), + ); + const connect = Effect.fn("ConnectionDriver.connect")(function* ( entry: ConnectionCatalogEntry, reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect, @@ -203,7 +216,7 @@ export const make = Effect.gen(function* () { Effect.withSpan("ConnectionDriver.preflight"), ); - return ConnectionDriver.of({ connect, checkRoute, preflight }); + return ConnectionDriver.of({ connect, checkRoute, preflight, checkHost }); }); export const layer = Layer.effect(ConnectionDriver, make); diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 83b28d7b9c904..bd6bfe971ed63 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -413,6 +413,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( Stream.die(new Error("Config is not used by registry tests.")), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Deferred.await(closed), } satisfies RpcSession.RpcSession), () => Ref.update(releasedSessions, (count) => count + 1), @@ -432,6 +433,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( checkRoute: (_entry, route) => checkRoute(route), preflight: (_entry, route) => checkRoute(route).pipe(Effect.map((check) => check === "answered")), + checkHost: () => Effect.succeed("unchecked"), }); const layerCache = Layer.succeed(Persistence.EnvironmentCacheStore, cacheStore); diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index cfdf8f633690e..480e27a9d7354 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -98,6 +98,16 @@ const LAN_THEN_RELAY_ENTRY: ConnectionCatalogEntry = { enabled: true, }; +const LAN_ENTRY: ConnectionCatalogEntry = { + target: LAN_ROUTE.target, + profile: LAN_ROUTE.profile, + enabled: true, +}; + +const PROBE_CAPABLE_CONFIG = { + environment: { capabilities: { connectionProbe: true } }, +} as ServerConfig; + function preparedFor(target: ConnectionTarget): PreparedConnection { return { ...PREPARED_CONNECTION, target }; } @@ -137,6 +147,8 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: readonly probe?: (attempt: number) => Effect.Effect; readonly initialConfig?: (attempt: number) => Effect.Effect; readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect; + readonly checkHost?: (prepared: PreparedConnection) => Effect.Effect; + readonly inboundFrames?: Effect.Effect; }) { const networkStatus = yield* SubscriptionRef.make( options?.networkStatus ?? "online", @@ -191,6 +203,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input), ready: options?.ready?.(attempt) ?? Effect.void, probe: options?.probe?.(attempt) ?? Effect.void, + inboundFrames: options?.inboundFrames ?? Effect.succeed(0), closed: Deferred.await(closed), } satisfies RpcSession.RpcSession), () => Ref.update(releaseCount, (count) => count + 1), @@ -233,6 +246,9 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: checkRoute: (_entry, route) => checkRoute(route), preflight: (_entry, route) => checkRoute(route).pipe(Effect.map((check) => check === "answered")), + checkHost: (prepared) => + options?.checkHost?.(prepared) ?? + Effect.succeed("unchecked"), }), ), ); @@ -540,10 +556,15 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); - it.effect("restarts the retry ladder when a long resume replaces a connected session", () => + it.effect("restarts the retry ladder when a resume replaces a silent socket", () => Effect.gen(function* () { - const harness = yield* makeHarness(); - const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: (attempt) => (attempt === 2 ? Effect.never : Effect.void), + checkHost: () => Effect.succeed("answered"), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); @@ -559,7 +580,8 @@ describe("EnvironmentSupervisor", () => { (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 2, ); - yield* harness.wake("application-active-reconnect"); + yield* harness.wake("application-active-long"); + yield* TestClock.adjust("500 millis"); yield* awaitState( supervisor.state, (state) => state.phase === "connected" && state.generation === 3 && state.attempt === 1, @@ -588,7 +610,7 @@ describe("EnvironmentSupervisor", () => { (state) => state.phase === "connecting" && state.attempt === 2, ); - yield* harness.wake("application-active-reconnect"); + yield* harness.wake("application-active-long"); yield* awaitState( supervisor.state, (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1, @@ -621,7 +643,7 @@ describe("EnvironmentSupervisor", () => { }), ); - it.effect("immediately replaces a mobile session after a long background resume", () => + it.effect("keeps a live mobile session after a long background resume", () => Effect.gen(function* () { const probeCount = yield* Ref.make(0); const harness = yield* makeHarness({ @@ -631,46 +653,336 @@ describe("EnvironmentSupervisor", () => { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-long"); + yield* TestClock.adjust("5 seconds"); + + expect(yield* Ref.get(probeCount)).toBe(1); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({ + phase: "connected", + generation: 1, + }); + }).pipe(Effect.provide(TestClock.layer())), + ); + + const makeSilentSocketHarness = (options?: { + readonly probe?: (attempt: number) => Effect.Effect; + readonly checkHost?: (check: number) => ConnectionDriver.RouteCheck; + }) => + Effect.gen(function* () { + const hostChecks = yield* Ref.make>([]); + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: options?.probe ?? ((attempt) => (attempt === 1 ? Effect.never : Effect.void)), + checkHost: (prepared) => + Ref.updateAndGet(hostChecks, (checks) => [...checks, prepared.target._tag]).pipe( + Effect.map((checks) => options?.checkHost?.(checks.length) ?? "answered"), + ), + }); + return { ...harness, hostChecks }; + }); + + it.effect.each(["application-active-probe", "application-active-long"] as const)( + "replaces a LAN socket that stays silent while its host answers afresh (%s)", + (wakeup) => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake(wakeup); + yield* TestClock.adjust("499 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* Ref.get(harness.hostChecks)).toEqual(["BearerConnectionTarget"]); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("replaces a remote socket only after 2.5 seconds of silence", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("2499 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); yield* awaitState( supervisor.state, - (state) => state.phase === "connected" && state.generation === 1, + (state) => state.phase === "connected" && state.generation === 2, ); - yield* harness.wake("application-active-reconnect"); + expect(yield* Ref.get(harness.hostChecks)).toEqual(["RelayConnectionTarget"]); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("replaces a silent socket even while the network reports offline", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.setNetworkStatus("offline"); + yield* TestClock.adjust("500 millis"); yield* awaitState( supervisor.state, (state) => state.phase === "connected" && state.generation === 2, ); + }).pipe(Effect.provide(TestClock.layer())), + ); - expect(yield* Ref.get(probeCount)).toBe(0); - expect(yield* Ref.get(harness.sessionCount)).toBe(2); - expect(yield* Ref.get(harness.releaseCount)).toBe(1); - }), + it.effect("replaces a remote socket after a long resume while the network reports offline", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.setNetworkStatus("offline"); + yield* harness.wake("application-active-long"); + yield* TestClock.adjust("2500 millis"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + }).pipe(Effect.provide(TestClock.layer())), ); - it.effect("replaces a mobile session when a long resume interrupts an active probe", () => + it.effect("checks the host afresh when the network changes after a frame", () => Effect.gen(function* () { + const frames = yield* Ref.make(0); const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + inboundFrames: Ref.get(frames), + checkHost: () => Effect.succeed("answered"), }); - const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("1 second"); + yield* Ref.update(frames, (count) => count + 1); + yield* TestClock.adjust("1 second"); + yield* harness.wake("network-changed"); + // The running check sees the frame at 2.5 s, so a fresh one starts then. + yield* TestClock.adjust("2999 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); yield* awaitState( supervisor.state, - (state) => state.phase === "connected" && state.generation === 1, + (state) => state.phase === "connected" && state.generation === 2, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("keeps a running check's window when the network changes during a long resume", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-long"); + yield* TestClock.adjust("1 second"); + yield* harness.wake("network-changed"); + yield* harness.setNetworkStatus("offline"); + yield* TestClock.adjust("1500 millis"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("checks the host again when one check goes unanswered", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness({ + checkHost: (check) => (check === 1 ? "silent" : "answered"), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("1499 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* Ref.get(harness.hostChecks)).toHaveLength(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("leaves a web or desktop foreground probe to its own deadline", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness({ + probe: (attempt) => (attempt === 1 ? Effect.sleep("4 seconds") : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active"); + yield* TestClock.adjust("20 seconds"); + + expect(yield* Ref.get(harness.hostChecks)).toEqual([]); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("checks the host when a network change arrives during a foreground probe", () => + Effect.gen(function* () { + const harness = yield* makeSilentSocketHarness(); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active"); + yield* TestClock.adjust("100 millis"); + yield* harness.wake("network-changed"); + yield* TestClock.adjust("499 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("keeps waiting for the probe when a frame arrives during the host check", () => + Effect.gen(function* () { + const frames = yield* Ref.make(0); + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: (attempt) => (attempt === 1 ? Effect.sleep("2 seconds") : Effect.void), + inboundFrames: Ref.get(frames), + checkHost: () => Ref.update(frames, (count) => count + 1).pipe(Effect.as("answered")), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); yield* harness.wake("application-active-probe"); - yield* Effect.yieldNow; - yield* harness.wake("application-active-reconnect"); + yield* TestClock.adjust("5 seconds"); + + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + expect((yield* SubscriptionRef.get(supervisor.state)).generation).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect.each([ + { name: "the host does not answer a fresh request", check: "silent", capable: true }, + { + name: "an older host answers probes with its full config", + check: "answered", + capable: false, + }, + ] as const)("waits out a LAN resume probe when $name", ({ check, capable }) => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => + Effect.succeed( + capable + ? PROBE_CAPABLE_CONFIG + : ({ environment: { capabilities: {} } } as ServerConfig), + ), + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + checkHost: () => Effect.succeed(check), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("2999 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); yield* awaitState( supervisor.state, (state) => state.phase === "connected" && state.generation === 2, ); + }).pipe(Effect.provide(TestClock.layer())), + ); - expect(yield* Ref.get(harness.sessionCount)).toBe(2); - expect(yield* Ref.get(harness.releaseCount)).toBe(1); - }), + it.effect.each(["application-active-probe", "network-changed"] as const)( + "gives a live remote socket 15 seconds to answer a %s probe", + (wakeup) => + Effect.gen(function* () { + // A slow path delays the fresh request as much as the probe. + const harness = yield* makeSilentSocketHarness({ + probe: (attempt) => (attempt === 1 ? Effect.sleep("4 seconds") : Effect.void), + checkHost: () => "silent", + }); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake(wakeup); + yield* TestClock.adjust("20 seconds"); + + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + expect((yield* SubscriptionRef.get(supervisor.state)).generation).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("replaces an unanswered long resume probe even while the network reports offline", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.wake("application-active-long"); + yield* harness.setNetworkStatus("offline"); + yield* TestClock.adjust("2999 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* TestClock.adjust("1 millis"); + const reconnected = yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(reconnected.lastFailure).toBeNull(); + }).pipe(Effect.provide(TestClock.layer())), ); it.effect("reconnects immediately when the session closes during a resume probe", () => diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d500fe6d8347c..c189874aaeecd 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -34,6 +34,7 @@ import { connectionRouteId, connectionRouteKind, connectionRoutes, + type ConnectionRouteKind, entryWithRoutes, type ReportedEndpoint, } from "./routes.ts"; @@ -47,6 +48,15 @@ const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_T const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds"); const CONNECTION_PROBE_TIMEOUT = "15 seconds"; const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds"; +// A probe still unanswered this long checks the host with a fresh request. If +// the host answers while the socket stays silent, the socket is dead. +const SILENT_SOCKET_CHECK_AFTER = "500 millis"; +const SILENT_SOCKET_RECHECK_EVERY = "1 second"; +// Off the LAN one lost packet can stall a live socket for a second, so a remote +// socket must stay silent this long. It must end before +// QUICK_CONNECTION_PROBE_TIMEOUT so a long resume replaces a silent socket +// instead of timing out. +const REMOTE_SILENT_SOCKET_AFTER = "2500 millis"; const BACKOFF_RESET_AFTER_MS = 30_000; const BETTER_ROUTE_CHECK_INTERVAL = "60 seconds"; // A better route that answered the check but then failed to connect is not @@ -521,7 +531,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( case "BetterRouteAvailable": break; case "Wakeup": - if (next.reason === "application-active-reconnect") { + // An attempt that ran into a long suspension is likely stuck. + if (next.reason === "application-active-long") { return true; } if (next.reason === "credentials-changed" && usesRelay) { @@ -547,24 +558,35 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( yield* Ref.set(preferredRouteId, Option.some(next.routeId)); return "reset" as const; } - if (next._tag !== "Wakeup") { - return undefined; - } - if (next.reason === "application-active-reconnect") { - // Mobile operating systems often kill a suspended socket without a close - // event. A probe would show a dead socket as "Resuming" until it times - // out, so a long background resume replaces the session at once. - return "reset" as const; - } // Only a session over T3 Connect holds the old account's credential. - if (next.reason === "credentials-changed" && isRelayLease(lease)) { + if (next._tag === "Wakeup" && next.reason === "credentials-changed" && isRelayLease(lease)) { yield* logManagedRelayAccountChange; return "end" as const; } return undefined; }); - const probeTimeoutFor = (next: SupervisorSignal): Duration.Input | undefined => { + const leaseRouteKind = (lease: ConnectionDriver.EnvironmentConnectionLease) => + currentRoutes.pipe( + Effect.map((routes) => { + const route = routes.find( + (candidate) => + connectionRouteId(candidate.target) === connectionRouteId(lease.prepared.target), + ); + return route === undefined ? undefined : connectionRouteKind(route); + }), + ); + + const isLongResume = (signal: SupervisorSignal) => + signal._tag === "Wakeup" && signal.reason === "application-active-long"; + + const answersQuickly = (routeKind: ConnectionRouteKind | undefined) => + routeKind === "loopback" || routeKind === "lan"; + + const probeTimeoutFor = ( + next: SupervisorSignal, + routeKind: ConnectionRouteKind | undefined, + ): Duration.Input | undefined => { switch (next._tag) { case "RetryRequested": return QUICK_CONNECTION_PROBE_TIMEOUT; @@ -574,9 +596,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (next.reason === "application-active") { return CONNECTION_PROBE_TIMEOUT; } - return next.reason === "application-active-probe" || next.reason === "network-changed" + // After a long suspension the data on screen is stale, so a slow + // answer reconnects as quickly as a missing one, even on a remote + // route where a reconnect takes longer than the wait it replaces. + if (next.reason === "application-active-long") { + return QUICK_CONNECTION_PROBE_TIMEOUT; + } + if (next.reason !== "application-active-probe" && next.reason !== "network-changed") { + return undefined; + } + // Only loopback and LAN answer within 3 seconds reliably. A tailnet + // over DERP, a VPN, or the relay can spike past it on a live socket; + // a dead one is still caught early by the silent socket check. + return answersQuickly(routeKind) ? QUICK_CONNECTION_PROBE_TIMEOUT - : undefined; + : CONNECTION_PROBE_TIMEOUT; case "ConnectRequested": case "DisconnectRequested": case "BetterRouteAvailable": @@ -584,6 +618,42 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } }; + // Ends "silent" when the host answers a fresh request while the socket + // delivers nothing, as when iOS kills a suspended socket without a close + // event, and "alive" once the socket delivers a frame. A slow or + // unreachable host never ends it. + const awaitSilentSocket = Effect.fn("EnvironmentSupervisor.awaitSilentSocket")(function* ( + lease: ConnectionDriver.EnvironmentConnectionLease, + routeKind: ConnectionRouteKind | undefined, + ) { + const startedAt = yield* Clock.monotonicTimeNanos; + yield* Effect.sleep(SILENT_SOCKET_CHECK_AFTER); + // An older host answers probes with its full config, which can be slow + // enough to lose to a fresh request on a healthy socket. + const answersCheapProbe = yield* lease.session.initialConfig.pipe( + Effect.map((config) => config.environment.capabilities.connectionProbe === true), + Effect.catchCause(() => Effect.succeed(false)), + ); + if (!answersCheapProbe) return yield* Effect.never; + // Frames that queued while the app was suspended have drained by now, so + // a later frame shows the path still works. + const framesBefore = yield* lease.session.inboundFrames; + const socketSilent = lease.session.inboundFrames.pipe( + Effect.map((frames) => frames === framesBefore), + ); + // A radio still waking up or a stale pooled connection can fail one check. + while ((yield* driver.checkHost(lease.prepared)) !== "answered") { + if (!(yield* socketSilent)) return "alive" as const; + yield* Effect.sleep(SILENT_SOCKET_RECHECK_EVERY); + } + if (!answersQuickly(routeKind)) { + const silentFor = (yield* Clock.monotonicTimeNanos) - startedAt; + const remaining = Duration.toNanosUnsafe(REMOTE_SILENT_SOCKET_AFTER) - silentFor; + if (remaining > 0n) yield* Effect.sleep(Duration.nanos(remaining)); + } + return (yield* socketSilent) ? ("silent" as const) : ("alive" as const); + }); + // Returning to the app, an explicit retry, and the network reporting offline // all probe the live session instead of replacing it, so a healthy socket is // not torn down (the offline report is often wrong, for example for a @@ -605,6 +675,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Stream.runForEach(() => requestBetterRouteCheck(lease)), Effect.forkScoped, ); + const routeKind = yield* leaseRouteKind(lease); for (;;) { const next = yield* takeSignal; const end = yield* connectedLeaseEnd(next, lease); @@ -617,12 +688,38 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ) { yield* requestBetterRouteCheck(lease); } - const probeTimeout = probeTimeoutFor(next); + const probeTimeout = probeTimeoutFor(next, routeKind); if (probeTimeout === undefined) { continue; } yield* Ref.set(probeUnanswered, true); const probe = yield* Effect.forkChild(lease.session.probe); + let silentSocket: Fiber.Fiber<"silent" | "alive"> | undefined; + // A signal during a running check, such as a network change that kills + // the socket, needs a fresh check only if the running one sees a frame. + let recheck = false; + let longResume = isLongResume(next); + const startSilentSocketCheck = Effect.gen(function* () { + silentSocket = yield* Effect.forkChild(awaitSilentSocket(lease, routeKind)); + recheck = false; + }); + // The web and desktop foreground probe allows a busy host 15 seconds, + // and a busy host can answer a fresh request before a queued probe. + const armSilentSocketCheck = Effect.fnUntraced(function* (signal: SupervisorSignal) { + if (signal._tag === "Wakeup" && signal.reason === "application-active") return; + if (silentSocket !== undefined) { + recheck = true; + return; + } + yield* startSilentSocketCheck; + }); + yield* armSilentSocketCheck(next); + const stopProbe = Effect.suspend(() => + Effect.all( + [Fiber.interrupt(probe), silentSocket ? Fiber.interrupt(silentSocket) : Effect.void], + { discard: true }, + ), + ); let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout); for (;;) { const remaining = deadline - (yield* Clock.monotonicTimeNanos); @@ -634,15 +731,37 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Effect.sleep(Duration.nanos(remaining > 0n ? remaining : 0n)).pipe( Effect.as({ _tag: "TimedOut" as const }), ), + (silentSocket ? Fiber.join(silentSocket) : Effect.never).pipe( + Effect.map((check) => + check === "silent" + ? { _tag: "SocketSilent" as const } + : { _tag: "SocketAlive" as const }, + ), + ), ]); + if (probeEvent._tag === "SocketAlive") { + silentSocket = undefined; + if (recheck) yield* startSilentSocketCheck; + continue; + } + if (probeEvent._tag === "SocketSilent") { + // The host is reachable, so replace the socket at once, even while + // the network reports offline. + yield* stopProbe; + return true; + } if (probeEvent._tag === "TimedOut") { - yield* Fiber.interrupt(probe); + yield* stopProbe; + // A long suspension leaves the socket suspect even when the network + // reports offline, so it is replaced like a silent one. + if (longResume) return true; return yield* new ConnectionTransientError({ reason: "timeout", detail: `${target.label} did not respond to a connection health check.`, }); } if (probeEvent._tag === "ProbeCompleted") { + if (silentSocket) yield* Fiber.interrupt(silentSocket); if (Exit.isSuccess(probeEvent.exit)) { yield* Ref.set(probeUnanswered, false); } @@ -651,11 +770,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease); if (endDuringProbe !== undefined) { - yield* Fiber.interrupt(probe); + yield* stopProbe; return endDuringProbe === "reset"; } - const signalTimeout = probeTimeoutFor(probeEvent.signal); + const signalTimeout = probeTimeoutFor(probeEvent.signal, routeKind); if (signalTimeout !== undefined) { + yield* armSilentSocketCheck(probeEvent.signal); + longResume ||= isLongResume(probeEvent.signal); const signalDeadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout); if (signalDeadline < deadline) deadline = signalDeadline; diff --git a/packages/client-runtime/src/connection/wakeups.ts b/packages/client-runtime/src/connection/wakeups.ts index b4315ad3a732a..c407c1b3a7dd9 100644 --- a/packages/client-runtime/src/connection/wakeups.ts +++ b/packages/client-runtime/src/connection/wakeups.ts @@ -5,7 +5,7 @@ import type * as Stream from "effect/Stream"; export type ConnectionWakeup = | "application-active" | "application-active-probe" - | "application-active-reconnect" + | "application-active-long" | "credentials-changed" | "network-changed"; @@ -13,7 +13,7 @@ function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean { return ( reason === "application-active" || reason === "application-active-probe" || - reason === "application-active-reconnect" + reason === "application-active-long" ); } @@ -21,9 +21,8 @@ export function resetsRetryBackoff(reason: ConnectionWakeup): boolean { return isApplicationActiveWakeup(reason) || reason === "network-changed"; } -// A long resume replaces the session, and the new session subscribes on its own. export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean { - return reason === "application-active" || reason === "application-active-probe"; + return isApplicationActiveWakeup(reason); } export class ConnectionWakeups extends Context.Service< diff --git a/packages/client-runtime/src/operations/commands.performance.test.ts b/packages/client-runtime/src/operations/commands.performance.test.ts index 279c994eb33ed..2378407c2df73 100644 --- a/packages/client-runtime/src/operations/commands.performance.test.ts +++ b/packages/client-runtime/src/operations/commands.performance.test.ts @@ -177,6 +177,7 @@ const makeSupervisor = Effect.fn("CommandPerformance.makeSupervisor")(function* subscribeServerConfig: (requestInput) => client.subscribeServerConfig(requestInput), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index bcfae19403483..04c6a55cf9f46 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -123,6 +123,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 78ad44aca19e2..4464510245d5e 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -51,6 +51,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index b0c5857ca9caf..3d4479db0f67e 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -346,6 +346,23 @@ describe("RpcSessionFactory", () => { }), ); + it.effect("counts every inbound frame, including ones that answer no request", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + const before = yield* session.inboundFrames; + + socket.serverMessage(encodeJson({ _tag: "Pong" })); + + expect(yield* session.inboundFrames).toBe(before + 1); + }), + ); + it.effect("closes the websocket when the session scope is released", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index ebd05d0b2d301..cb15e42b93b0f 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -53,6 +53,8 @@ export interface RpcSession { ) => ServerConfigSubscription; readonly ready: Effect.Effect; readonly probe: Effect.Effect; + /** Frames received so far: proof the socket is alive that needs no reply. */ + readonly inboundFrames: Effect.Effect; readonly closed: Effect.Effect; } @@ -211,9 +213,21 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Effect.asVoid, ), }); + // The protocol never reopens its socket, so every frame counted here came + // from this session's one socket. + let inboundFrames = 0; + const countingWebSocketConstructor: typeof webSocketConstructor = (url, options) => { + const socket = webSocketConstructor(url, options); + socket.addEventListener("message", () => { + inboundFrames += 1; + }); + return socket; + }; const layerSocket = Socket.layerWebSocket(connection.socketUrl, { openTimeout: SOCKET_OPEN_TIMEOUT, - }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor))); + }).pipe( + Layer.provide(Layer.succeed(Socket.WebSocketConstructor, countingWebSocketConstructor)), + ); const layerProtocol = Layer.effect( RpcClient.Protocol, RpcClient.makeProtocolSocket({ @@ -417,6 +431,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Effect.raceFirst(Deferred.await(disconnected)), ), probe, + inboundFrames: Effect.sync(() => inboundFrames), closed: Effect.raceFirst( Deferred.await(disconnected), Deferred.await(configSubscriptionClosed), diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 474ef49de53dc..966c68dabd9e3 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -513,6 +513,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 9ee4e55697faa..b4f090d9b2fbe 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -88,6 +88,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index 21ad521badce0..247f4c650ae71 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -135,6 +135,7 @@ const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* ( subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 2941e81278748..9d1df33d1f70b 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -60,6 +60,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index b086612aa4515..43066806ec74b 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -100,6 +100,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 93bee9f46dd09..3c1c153621e28 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -94,6 +94,7 @@ function testSession( subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index 0c839e188e106..0a8b6f9cb6b59 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -94,6 +94,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/vcsAction.test.ts b/packages/client-runtime/src/state/vcsAction.test.ts index 52fc5575d04ab..2dfc43e91ee8a 100644 --- a/packages/client-runtime/src/state/vcsAction.test.ts +++ b/packages/client-runtime/src/state/vcsAction.test.ts @@ -89,6 +89,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + inboundFrames: Effect.succeed(0), closed: Effect.never, }; } From 55cf7a41c090799911074b41e42c4cb69748a8ef Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 27 Oct 2026 01:32:06 -0700 Subject: [PATCH 2/2] fix(client-runtime): host checks bypass the HTTP cache Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/connection/driver.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index 7a08368096e58..57e3a608258d4 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -4,6 +4,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Scope from "effect/Scope"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as HttpClient from "effect/http/HttpClient"; import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts"; @@ -155,6 +156,8 @@ export const make = Effect.gen(function* () { descriptor.environmentId === environmentId ? "answered" : "silent", ), Effect.orElseSucceed((): RouteCheck => "silent"), + // A cached descriptor would answer for a host that cannot be reached. + Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }), Effect.provideService(HttpClient.HttpClient, httpClient), );