From 0a649f9827ee05b94375b4ddad528d5545954fdb Mon Sep 17 00:00:00 2001 From: levelorbit <172493156+levelorbit@users.noreply.github.com> Date: Wed, 4 Nov 2026 14:03:08 +0100 Subject: [PATCH 1/4] packages/client-runtime/connection: open the replacement before closing the old session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The supervisor ended a session before it started the next one, on resume, on network changes and on route upgrades. On a remote route a resume waited out the 2.5 s silence window before any new connection began, and every switch showed "Syncing threads…" even when the host was reachable. Fixes #30833. --- docs/internals/connection-runtime.md | 45 +- .../client-runtime/src/connection/driver.ts | 2 +- .../src/connection/supervisor.test.ts | 462 ++++++++++++++++- .../src/connection/supervisor.ts | 488 ++++++++++++++---- 4 files changed, 867 insertions(+), 130 deletions(-) diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 698ddac7a73f9..961bdb261a5d7 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -17,22 +17,35 @@ 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, 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 established session instead of tearing it down. 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, a replacement +session starts connecting. The socket itself is declared dead, even while the +network reports offline, immediately 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. A long suspension +also restarts a pending replacement, which may have stalled with the app. + +A replacement that connects first takes over the session directly, so +subscriptions move without the client showing a reconnect, unless it landed on a +worse route than the live socket; then it waits as the fallback. One still +connecting when the socket is declared dead becomes the reconnect attempt; one +that is not an upgrade closes when the live socket proves alive. A network +change restarts or starts a replacement at once, except on loopback, because it +can strand the socket's route, for example when leaving a LAN, and anything +opened over the old path; that costs one extra connection per environment per +change. A better route connects beside the live session the same way and takes +over when it connects on that route; if it fails, the live session stays and the +route waits five minutes. 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 6039f25fa99d6..d644e62d41025 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -47,7 +47,7 @@ export type RoutePreflight = "ready" | "unprepared" | "silent" | "unchecked"; const ROUTE_CHECK_TIMEOUT_MS = 2_500; // Preferred direct routes share one head start, so a relay connection does not // wait the full probe timeout for a LAN the device has already left. A route -// that misses it connects over relay, then upgrades with a second reconnect. +// that misses it connects over relay, then upgrades over a second connection. const ROUTE_CHECK_HEAD_START_MS = 500; export class ConnectionDriver extends Context.Service< diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index b4fc6955366a0..80cb89448d98b 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -9,6 +9,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Random from "effect/Random"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; @@ -142,6 +143,26 @@ function awaitState( ); } +// Records what the UI would show, to prove a session moved without +// passing through connecting or a missing session. +function recordTransitions(supervisor: EnvironmentSupervisor.EnvironmentSupervisor["Service"]) { + return Effect.gen(function* () { + const phases = yield* Ref.make>([]); + const sessions = yield* Ref.make>([]); + yield* SubscriptionRef.changes(supervisor.state).pipe( + Stream.runForEach((state) => Ref.update(phases, (all) => [...all, state.phase])), + Effect.forkChild({ startImmediately: true }), + ); + yield* SubscriptionRef.changes(supervisor.session).pipe( + Stream.runForEach((session) => + Ref.update(sessions, (all) => [...all, Option.isSome(session)]), + ), + Effect.forkChild({ startImmediately: true }), + ); + return { phases: Ref.get(phases), sessions: Ref.get(sessions) }; + }); +} + const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: { readonly networkStatus?: NetworkStatus; readonly prepare?: ( @@ -732,13 +753,19 @@ describe("EnvironmentSupervisor", () => { ); const makeSilentSocketHarness = (options?: { + readonly prepare?: ( + attempt: number, + target: ConnectionTarget, + ) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; readonly checkHost?: (check: number) => ConnectionDriver.RouteCheck; + readonly inboundFrames?: Effect.Effect; }) => Effect.gen(function* () { const hostChecks = yield* Ref.make>([]); const harness = yield* makeHarness({ - prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + prepare: options?.prepare ?? ((_attempt, target) => Effect.succeed(preparedFor(target))), + ...(options?.inboundFrames === undefined ? {} : { inboundFrames: options.inboundFrames }), initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), probe: options?.probe ?? ((attempt) => (attempt === 1 ? Effect.never : Effect.void)), checkHost: (prepared) => @@ -773,16 +800,24 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); - it.effect("replaces a remote socket only after 2.5 seconds of silence", () => + it.effect("moves a silent remote socket to a replacement opened when its host answers", () => Effect.gen(function* () { - const harness = yield* makeSilentSocketHarness(); + const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + Effect.sleep(attempt === 1 ? "0 millis" : "1 second").pipe( + Effect.as(preparedFor(target)), + ), + }); const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + const transitions = yield* recordTransitions(supervisor); yield* harness.wake("application-active-probe"); - yield* TestClock.adjust("2499 millis"); + // The host answers at 500 ms; the replacement connects a second later, + // before the socket's 2.5 second silence window ends. + yield* TestClock.adjust("1499 millis"); expect(yield* Ref.get(harness.sessionCount)).toBe(1); yield* TestClock.adjust("1 millis"); @@ -790,7 +825,15 @@ describe("EnvironmentSupervisor", () => { supervisor.state, (state) => state.phase === "connected" && state.generation === 2, ); + expect(yield* transitions.phases).not.toContain("connecting"); + // The current session, then the replacement, each published once. + expect(yield* transitions.sessions).toEqual([true, true]); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); expect(yield* Ref.get(harness.hostChecks)).toEqual(["RelayConnectionTarget"]); + + yield* supervisor.disconnect; + yield* awaitState(supervisor.state, (state) => state.phase === "available"); + expect(yield* Ref.get(harness.releaseCount)).toBe(2); }).pipe(Effect.provide(TestClock.layer())), ); @@ -855,12 +898,11 @@ describe("EnvironmentSupervisor", () => { 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), + // Replacements never connect, so only the socket's own checks decide. + const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + attempt === 1 ? Effect.succeed(preparedFor(target)) : Effect.never, inboundFrames: Ref.get(frames), - checkHost: () => Effect.succeed("answered"), }); const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { initiallyDesired: true, @@ -874,13 +916,13 @@ describe("EnvironmentSupervisor", () => { 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); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected"); yield* TestClock.adjust("1 millis"); - yield* awaitState( - supervisor.state, - (state) => state.phase === "connected" && state.generation === 2, - ); + yield* awaitState(supervisor.state, (state) => state.phase === "connecting"); + // The network change restarted the first replacement, the frame dropped + // the second, and the third becomes the attempt. + expect(yield* Ref.get(harness.prepareCount)).toBe(4); }).pipe(Effect.provide(TestClock.layer())), ); @@ -947,7 +989,11 @@ describe("EnvironmentSupervisor", () => { it.effect("checks the host when a network change arrives during a foreground probe", () => Effect.gen(function* () { - const harness = yield* makeSilentSocketHarness(); + // Replacements never connect, so only the socket's own checks decide. + const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + attempt === 1 ? Effect.succeed(preparedFor(target)) : Effect.never, + }); const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); @@ -957,13 +1003,15 @@ describe("EnvironmentSupervisor", () => { yield* TestClock.adjust("100 millis"); yield* harness.wake("network-changed"); yield* TestClock.adjust("499 millis"); - expect(yield* Ref.get(harness.sessionCount)).toBe(1); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected"); yield* TestClock.adjust("1 millis"); + // The network change opened the replacement, which becomes the attempt. yield* awaitState( supervisor.state, - (state) => state.phase === "connected" && state.generation === 2, + (state) => state.phase === "connecting" && state.attempt === 1 && state.generation === 2, ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); }).pipe(Effect.provide(TestClock.layer())), ); @@ -1019,8 +1067,11 @@ describe("EnvironmentSupervisor", () => { "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. + // A slow path delays the fresh request as much as the probe, and the + // network change's replacement never connects. const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + attempt === 1 ? Effect.succeed(preparedFor(target)) : Effect.never, probe: (attempt) => (attempt === 1 ? Effect.sleep("4 seconds") : Effect.void), checkHost: () => "silent", }); @@ -1037,6 +1088,159 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("keeps a live socket and closes its replacement when it answers a network change", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: (attempt) => (attempt === 1 ? Effect.sleep("1 second") : Effect.void), + ready: (attempt) => (attempt === 2 ? Effect.never : Effect.void), + }); + 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("network-changed"); + yield* TestClock.adjust("5 seconds"); + + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({ + phase: "connected", + generation: 1, + }); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("adopts a pending replacement when the old session closes", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (attempt, target) => + Effect.sleep(attempt === 1 ? "0 millis" : "1 second").pipe( + Effect.as(preparedFor(target)), + ), + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + 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("network-changed"); + yield* TestClock.adjust("100 millis"); + yield* harness.closeLatestSession(); + // The replacement started at 0 ms finishes at 1 s, before a fresh attempt could. + yield* TestClock.adjust("900 millis"); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("restarts a pending replacement after a long suspension", () => + Effect.gen(function* () { + // Replacements never connect and the host never answers afresh, so the + // long resume's probe deadline ends the session. + const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + attempt === 1 ? Effect.succeed(preparedFor(target)) : Effect.never, + 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("network-changed"); + yield* TestClock.adjust("1 second"); + yield* harness.wake("application-active-long"); + yield* TestClock.adjust("3 seconds"); + + yield* awaitState(supervisor.state, (state) => state.phase === "connecting"); + // The stalled replacement was restarted, and the restarted one becomes + // the attempt. + expect(yield* Ref.get(harness.prepareCount)).toBe(3); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("restarts a pending replacement when the network changes", () => + Effect.gen(function* () { + // The replacement opened over the old path stalls; a later one connects. + const harness = yield* makeSilentSocketHarness({ + prepare: (attempt, target) => + attempt === 2 ? Effect.never : Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + const transitions = yield* recordTransitions(supervisor); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("1 second"); + yield* harness.wake("network-changed"); + yield* TestClock.adjust("1 millis"); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* transitions.phases).not.toContain("connecting"); + expect(yield* Ref.get(harness.prepareCount)).toBe(3); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("starts afresh instead of adopting a replacement past its setup time", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (attempt, target) => + attempt === 1 ? Effect.succeed(preparedFor(target)) : Effect.never, + initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG), + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + 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("network-changed"); + // The relay probe times out at 15 s, when the replacement's single-route + // setup time is also spent. + yield* TestClock.adjust("15 seconds"); + + yield* awaitState(supervisor.state, (state) => state.phase === "connecting"); + expect(yield* Ref.get(harness.prepareCount)).toBe(3); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("closes a pending replacement on disconnect", () => + Effect.gen(function* () { + 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), + ready: (attempt) => (attempt === 2 ? Effect.never : Effect.void), + }); + 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("network-changed"); + yield* TestClock.adjust("1 second"); + yield* supervisor.disconnect; + yield* awaitState(supervisor.state, (state) => state.phase === "available"); + + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(2); + }).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({ @@ -1688,6 +1892,7 @@ describe("EnvironmentSupervisor routes", () => { "RelayConnectionTarget", ); + const transitions = yield* recordTransitions(supervisor); yield* Ref.set(lanReachable, true); yield* harness.setNetworkStatus("unknown"); yield* harness.setNetworkStatus("online"); @@ -1700,9 +1905,143 @@ describe("EnvironmentSupervisor routes", () => { "BearerConnectionTarget", ); expect(yield* Ref.get(harness.releaseCount)).toBe(1); + expect(yield* transitions.phases).not.toContain("connecting"); + expect(yield* transitions.sessions).not.toContain(false); }), ); + it.effect("keeps the relay session when a better LAN route answers but fails to connect", () => + Effect.gen(function* () { + const lanReachable = yield* Ref.make(false); + const lanTried = yield* Deferred.make(); + const lanPrepares = yield* Ref.make(0); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => + target._tag === "BearerConnectionTarget" + ? Ref.update(lanPrepares, (count) => count + 1).pipe( + Effect.andThen(Deferred.succeed(lanTried, undefined)), + Effect.andThen(Effect.fail(transient("LAN socket refused."))), + ) + : Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + const transitions = yield* recordTransitions(supervisor); + yield* Ref.set(lanReachable, true); + yield* harness.setNetworkStatus("unknown"); + yield* harness.setNetworkStatus("online"); + yield* Deferred.await(lanTried); + yield* TestClock.adjust("1 second"); + + // The upgrade fell back to relay, the live session's route, so it closed. + expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({ + phase: "connected", + generation: 1, + }); + expect(yield* transitions.phases).not.toContain("connecting"); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + + // The LAN route cools down for five minutes despite the minute checks. + yield* TestClock.adjust("4 minutes"); + expect(yield* Ref.get(lanPrepares)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("cools down a stalled LAN upgrade, then tries it again", () => + Effect.gen(function* () { + const lanReachable = yield* Ref.make(false); + const lanPrepares = yield* Queue.unbounded(); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => + target._tag === "BearerConnectionTarget" + ? Queue.offer(lanPrepares, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* Ref.set(lanReachable, true); + yield* harness.setNetworkStatus("unknown"); + yield* harness.setNetworkStatus("online"); + yield* Queue.take(lanPrepares); + // The next minute's check finds the upgrade past its 30 s setup time and + // cools the route down instead of opening another stalled connection. + yield* TestClock.adjust("1 minute"); + expect(yield* Queue.size(lanPrepares)).toBe(0); + // The five minute cooldown ends at the sixth minute's check. + yield* TestClock.adjust("5 minutes"); + yield* Queue.take(lanPrepares); + expect((yield* SubscriptionRef.get(supervisor.state)).generation).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("moves to a connecting LAN upgrade when the relay session closes", () => + Effect.gen(function* () { + const lanReachable = yield* Ref.make(false); + const lanStarted = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => + target._tag === "BearerConnectionTarget" + ? Deferred.succeed(lanStarted, undefined).pipe( + Effect.andThen(Effect.sleep("2 seconds")), + Effect.as(preparedFor(target)), + ) + : Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + const transitions = yield* recordTransitions(supervisor); + yield* Ref.set(lanReachable, true); + yield* harness.setNetworkStatus("unknown"); + yield* harness.setNetworkStatus("online"); + yield* Deferred.await(lanStarted); + expect(yield* transitions.phases).not.toContain("connecting"); + yield* harness.closeLatestSession(); + yield* TestClock.adjust("2 seconds"); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "BearerConnectionTarget", + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("checks a dead LAN route less often until the network changes", () => Effect.gen(function* () { const lanReachable = yield* Ref.make(false); @@ -1858,8 +2197,11 @@ describe("EnvironmentSupervisor routes", () => { "BearerConnectionTarget", ); + const transitions = yield* recordTransitions(supervisor); yield* Ref.set(lanReachable, false); yield* harness.wake("network-changed"); + // The replacement lands on relay, which ranks below the LAN, so it waits + // as the fallback until the LAN probe times out, then takes over. yield* TestClock.adjust("3 seconds"); yield* awaitState( @@ -1869,9 +2211,93 @@ describe("EnvironmentSupervisor routes", () => { expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( "RelayConnectionTarget", ); + expect(yield* transitions.phases).not.toContain("connecting"); + expect(yield* transitions.sessions).not.toContain(false); }), ); + it.effect.each([ + ["network-changed", "application-active-long"], + ["application-active-long", "network-changed"], + ] as const)("moves off a dead LAN socket on a resume that reports %s, then %s", (wakeups) => + Effect.gen(function* () { + const lanReachable = yield* Ref.make(true); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + probe: () => + Ref.get(lanReachable).pipe( + Effect.flatMap((reachable) => (reachable ? Effect.void : Effect.never)), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + const transitions = yield* recordTransitions(supervisor); + yield* Ref.set(lanReachable, false); + for (const wakeup of wakeups) yield* harness.wake(wakeup); + yield* TestClock.adjust("3 seconds"); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "RelayConnectionTarget", + ); + expect(yield* transitions.sessions).toEqual([true, true]); + }), + ); + + it.effect("keeps a live LAN socket when a network change replacement lands on relay", () => + Effect.gen(function* () { + const lanSlow = yield* Ref.make(false); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanSlow).pipe( + Effect.flatMap((slow) => Effect.sleep(slow ? "1 second" : "0 millis")), + Effect.as("answered" as const), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + probe: (attempt) => (attempt === 1 ? Effect.sleep("1500 millis") : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + // The LAN answers after the replacement's head start, so it lands on relay. + yield* Ref.set(lanSlow, true); + yield* harness.wake("network-changed"); + yield* TestClock.adjust("3 seconds"); + + expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({ + phase: "connected", + generation: 1, + }); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "BearerConnectionTarget", + ); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect.each(["initial", "late", "withdrawn", "withdrawn-upgrade", "omitted"] as const)( "learns the %s direct endpoint over T3 Connect", (discovery) => diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 5f279edf568c4..20f50c5515c7b 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -11,6 +11,7 @@ import * as Queue from "effect/Queue"; import * as Random from "effect/Random"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as Tracer from "effect/Tracer"; @@ -79,7 +80,8 @@ type SupervisorSignal = | { readonly _tag: "RetryRequested" } | { readonly _tag: "NetworkChanged"; readonly network: NetworkStatus } | { readonly _tag: "Wakeup"; readonly reason: ConnectionWakeups.ConnectionWakeup } - | { readonly _tag: "BetterRouteAvailable"; readonly routeId: string }; + | { readonly _tag: "BetterRouteAvailable"; readonly routeId: string } + | { readonly _tag: "ReplacementReady"; readonly replacement: Replacement }; interface PendingRetryTrace { readonly previousAttempt: Tracer.Span; @@ -97,12 +99,35 @@ interface TracedAttemptFailure { readonly attemptSpan: Option.Option; } +interface EstablishedLease { + readonly attemptSpan: Option.Option; + readonly lease: ConnectionDriver.EnvironmentConnectionLease; +} + +// A connection opened beside a live session, so the session can move to it +// without passing through disconnected. Its progress stays hidden until the +// old session ends and it becomes the visible attempt. +interface Replacement { + readonly fiber: Fiber.Fiber; + readonly scope: Scope.Closeable; + readonly startedAt: bigint; + readonly synchronizedRoutes: Ref.Ref; + readonly upgradeTo: Option.Option; + readonly reveal: Effect.Effect; +} + +interface Handoff { + readonly replacement: Replacement; + readonly ready: Option.Option; +} + type AttemptOutcome = | { readonly _tag: "Interrupted"; readonly established: boolean; readonly stable: boolean; readonly resetRetry: boolean; + readonly handoff?: Handoff; } | { readonly _tag: "Failure"; @@ -114,13 +139,7 @@ type AttemptOutcome = type EstablishmentEvent = | { readonly _tag: "Completed"; - readonly exit: Exit.Exit< - { - readonly attemptSpan: Option.Option; - readonly lease: ConnectionDriver.EnvironmentConnectionLease; - }, - TracedAttemptFailure - >; + readonly exit: Exit.Exit; } | { readonly _tag: "Interrupted"; readonly resetRetry: boolean } | { readonly _tag: "TimedOut" }; @@ -285,6 +304,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }`; yield* annotateTarget(target); + const supervisorScope = yield* Scope.Scope; const connectivity = yield* Connectivity.Connectivity; const driver = yield* ConnectionDriver.ConnectionDriver; const wakeups = yield* ConnectionWakeups.ConnectionWakeups; @@ -314,7 +334,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( // fresh by each attempt and check instead of being fixed at creation. const currentEntry = yield* Ref.make(entry); const currentRoutes = Ref.get(currentEntry).pipe(Effect.map(connectionRoutes)); - const preferredRouteId = yield* Ref.make(Option.none()); + const pendingReplacement = yield* Ref.make(Option.none()); + const replacementLock = yield* Semaphore.make(1); const routeCooldowns = yield* Ref.make>(new Map()); // The generation keeps a check that started before a reset from recording // its stale failures after it. @@ -327,13 +348,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( routes: new Map(), })); - const attemptEntry = Effect.gen(function* () { + const attemptEntry = Effect.fnUntraced(function* (upgradeTo: Option.Option) { const latest = yield* Ref.get(currentEntry); const routes = connectionRoutes(latest); - const preferred = yield* Ref.getAndSet(preferredRouteId, Option.none()); - if (Option.isNone(preferred)) return latest; + if (Option.isNone(upgradeTo)) return latest; const route = routes.find( - (candidate) => connectionRouteId(candidate.target) === preferred.value, + (candidate) => connectionRouteId(candidate.target) === upgradeTo.value, ); return route === undefined ? latest @@ -377,6 +397,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }), ); + const routeRank = (routeId: string) => + currentRoutes.pipe( + Effect.map((routes) => + routes.findIndex((route) => connectionRouteId(route.target) === routeId), + ), + ); + const checkBetterRoutes = Effect.fnUntraced(function* (lease: ConnectedLease) { const latest = yield* Ref.get(currentEntry); const current = yield* routeIndex(lease); @@ -428,6 +455,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( { discard: true }, ); + // Every session change restarts each subscription, so an unchanged session + // is not published again. + const setLease = Effect.fnUntraced(function* ( + lease: ConnectionDriver.EnvironmentConnectionLease, + ) { + const current = yield* SubscriptionRef.get(session); + if (Option.isSome(current) && current.value === lease.session) return; + yield* SubscriptionRef.set(prepared, Option.some(lease.prepared)); + yield* SubscriptionRef.set(session, Option.some(lease.session)); + }); + const setState = Effect.fn("EnvironmentSupervisor.setState")(function* ( next: SupervisorConnectionState, ) { @@ -461,18 +499,23 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ); }); - const establishConnection = Effect.fnUntraced(function* ( - attempt: number, - generation: number, - lastFailure: ConnectionAttemptError | null, - synchronizedRoutes: Ref.Ref, - ) { - return yield* driver.connect(yield* attemptEntry, (progress) => + type ProgressReporter = ( + progress: ConnectionDriver.ConnectionDriverProgress, + ) => Effect.Effect; + + const countingSynchronized = + (synchronizedRoutes: Ref.Ref, report: ProgressReporter): ProgressReporter => + (progress) => (progress.stage === "synchronizing" ? Ref.update(synchronizedRoutes, (count) => count + 1) : Effect.void - ).pipe(Effect.andThen(reportProgress(attempt, generation, lastFailure, progress))), - ); + ).pipe(Effect.andThen(report(progress))); + + const establishConnection = Effect.fnUntraced(function* ( + preferredRoute: Option.Option, + report: ProgressReporter, + ) { + return yield* driver.connect(yield* attemptEntry(preferredRoute), report); }); // Low-cardinality dimensions for time-to-connected dashboards; the error @@ -550,19 +593,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const establishTracedConnection = Effect.fnUntraced(function* ( attempt: number, generation: number, - lastFailure: ConnectionAttemptError | null, pendingRetry: Option.Option, - synchronizedRoutes: Ref.Ref, + preferredRoute: Option.Option, + report: ProgressReporter, ) { if (usesRelay) { return yield* traceRelayEstablishment( - establishConnection(attempt, generation, lastFailure, synchronizedRoutes), + establishConnection(preferredRoute, report), attempt, generation, pendingRetry, ); } - return yield* establishConnection(attempt, generation, lastFailure, synchronizedRoutes).pipe( + return yield* establishConnection(preferredRoute, report).pipe( Effect.map((lease) => ({ attemptSpan: Option.none(), lease, @@ -574,6 +617,160 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ); }); + // The fiber stops before its sockets close, so a discarded replacement ends + // interrupted instead of failed. + const closeReplacement = (replacement: Replacement) => + Fiber.interrupt(replacement.fiber).pipe( + Effect.ensuring(Scope.close(replacement.scope, Exit.void)), + ); + + // Whoever takes the pending replacement owns it: it either connects over it + // or closes it. + const discardReplacement = (when: (replacement: Replacement) => boolean = () => true) => + Effect.uninterruptible( + Ref.modify( + pendingReplacement, + (current): readonly [Option.Option, Option.Option] => + Option.isSome(current) && when(current.value) + ? [current, Option.none()] + : [Option.none(), current], + ).pipe( + Effect.tap((taken) => (Option.isSome(taken) ? closeReplacement(taken.value) : Effect.void)), + ), + ); + + const discardSuspectReplacement = discardReplacement((replacement) => + Option.isNone(replacement.upgradeTo), + ); + + const takeHandoff = Effect.gen(function* () { + const taken = yield* Ref.getAndSet(pendingReplacement, Option.none()); + if (Option.isNone(taken)) return Option.none(); + const replacement = taken.value; + const exit = replacement.fiber.pollUnsafe(); + if (exit !== undefined && Exit.isSuccess(exit)) { + return Option.some({ replacement, ready: Option.some(exit.value) }); + } + if (exit !== undefined || (yield* replacementExpired(replacement))) { + yield* closeReplacement(replacement); + return Option.none(); + } + return Option.some({ replacement, ready: Option.none() }); + }); + + const coolDownRoute = Effect.fnUntraced(function* (routeId: string) { + const until = (yield* Clock.monotonicTimeNanos) + BigInt(BETTER_ROUTE_COOLDOWN_MS) * 1_000_000n; + yield* Ref.update(routeCooldowns, (current) => new Map(current).set(routeId, until)); + }); + + const settleReplacement = Effect.fn("EnvironmentSupervisor.settleReplacement")(function* ( + replacement: Replacement, + exit: Exit.Exit, + ) { + if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) return; + const landedOn = Exit.isSuccess(exit) + ? connectionRouteId(exit.value.lease.prepared.target) + : undefined; + const usable = Option.match(replacement.upgradeTo, { + onNone: () => landedOn !== undefined, + onSome: (routeId) => routeId === landedOn, + }); + if (usable) { + return yield* signal({ _tag: "ReplacementReady", replacement }); + } + if (Option.isSome(replacement.upgradeTo)) yield* coolDownRoute(replacement.upgradeTo.value); + yield* discardReplacement((current) => current === replacement); + }); + + // Each route may use the full setup time before the next is tried, and + // each route that reaches synchronizing adds its own snapshot budget, so + // a route that fails late does not eat the next route's time. Untraced: it + // is a calculation rerun on every deadline wake. + const setupDeadline = Effect.fnUntraced(function* ( + startedAt: bigint, + synchronizedRoutes: Ref.Ref, + ) { + const routeCount = connectionRoutes(yield* Ref.get(currentEntry)).length; + const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); + return ( + startedAt + + Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routeCount)) + + synchronized * Duration.toNanosUnsafe(synchronizationTimeout) + ); + }); + + // A hidden replacement has no timer of its own; past its setup budget it + // is stale, so it is neither adopted nor allowed to block a new one. + const replacementExpired = (replacement: Replacement) => + Effect.all([ + setupDeadline(replacement.startedAt, replacement.synchronizedRoutes), + Clock.monotonicTimeNanos, + ]).pipe(Effect.map(([deadline, now]) => now >= deadline)); + + // `generation` is the live session's; the replacement reports progress as + // the first attempt of the next one, which it becomes if the session ends. + const startReplacement = Effect.fn("EnvironmentSupervisor.startReplacement")( + function* (generation: number, upgradeTo: Option.Option) { + const pending = yield* Ref.get(pendingReplacement); + if (Option.isSome(pending)) { + if (!(yield* replacementExpired(pending.value))) return; + yield* discardReplacement(); + // A better route that answered its check but stalled waits like one + // that failed, instead of a new stalled connection every minute. + const stalledUpgrade = pending.value.upgradeTo; + if (Option.isSome(stalledUpgrade)) { + yield* coolDownRoute(stalledUpgrade.value); + if (Option.contains(upgradeTo, stalledUpgrade.value)) return; + } + } + const startedAt = yield* Clock.monotonicTimeNanos; + const scope = yield* Scope.fork(supervisorScope); + const synchronizedRoutes = yield* Ref.make(0); + const lock = yield* Semaphore.make(1); + let visible = false; + let latest: ConnectionDriver.ConnectionDriverProgress | undefined; + const report: ProgressReporter = (progress) => + lock.withPermit( + Effect.suspend(() => { + latest = progress; + return visible ? reportProgress(1, generation + 1, null, progress) : Effect.void; + }), + ); + const reveal = lock.withPermit( + Effect.suspend(() => { + visible = true; + return latest === undefined + ? Effect.void + : reportProgress(1, generation + 1, null, latest); + }), + ); + const fiber = yield* establishTracedConnection( + 1, + generation + 1, + Option.none(), + upgradeTo, + countingSynchronized(synchronizedRoutes, report), + ).pipe(Scope.provide(scope), Effect.forkIn(scope)); + const replacement: Replacement = { + fiber, + scope, + startedAt, + synchronizedRoutes, + upgradeTo, + reveal, + }; + yield* Ref.set(pendingReplacement, Option.some(replacement)); + yield* Fiber.await(fiber).pipe( + Effect.flatMap((exit) => settleReplacement(replacement, exit)), + Effect.forkIn(supervisorScope), + ); + }, + // The lock makes check-then-set atomic for concurrent callers, and an + // interrupted caller cannot leave behind a scope that nothing owns. + (effect) => replacementLock.withPermit(effect), + Effect.uninterruptible, + ); + const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () { // A primary attempt keeps running through an offline report, so the // network returning restarts it in case it stalled meanwhile. @@ -597,6 +794,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( break; case "ConnectRequested": case "BetterRouteAvailable": + case "ReplacementReady": break; case "Wakeup": // An attempt that ran into a long suspension is likely stuck. @@ -617,19 +815,42 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const connectedLeaseEnd = Effect.fnUntraced(function* ( next: SupervisorSignal, - lease: ConnectionDriver.EnvironmentConnectionLease, - ) { + lease: ConnectedLease, + generation: number, + ): Effect.fn.Return<"end" | "replace" | undefined> { if (next._tag === "DisconnectRequested") { return "end" as const; } + // A check queued before the session moved can name a route that is no + // longer better than the live one. if (next._tag === "BetterRouteAvailable") { - yield* Ref.set(preferredRouteId, Option.some(next.routeId)); - return "reset" as const; + const rank = yield* routeRank(next.routeId); + if (rank === -1 || rank >= (yield* routeIndex(lease))) return undefined; + yield* discardSuspectReplacement; + yield* startReplacement(generation, Option.some(next.routeId)); + return undefined; } - // Only a session over T3 Connect holds the old account's credential. - if (next._tag === "Wakeup" && next.reason === "credentials-changed" && isRelayLease(lease)) { - yield* logManagedRelayAccountChange; - return "end" as const; + // The attempt that owns the lease takes the replacement once the monitor + // has stopped, so a close racing this signal cannot strand it. + if (next._tag === "ReplacementReady") { + const pending = yield* Ref.get(pendingReplacement); + if (Option.isNone(pending) || pending.value !== next.replacement) return undefined; + // One on a worse route waits as the fallback in case the socket is dead. + const exit = next.replacement.fiber.pollUnsafe(); + if (Option.isNone(next.replacement.upgradeTo) && exit !== undefined && Exit.isSuccess(exit)) { + const rank = yield* routeRank(connectionRouteId(exit.value.lease.prepared.target)); + if (rank > (yield* routeIndex(lease))) return undefined; + } + return "replace"; + } + if (next._tag === "Wakeup" && next.reason === "credentials-changed") { + // A replacement may have been prepared with the old account. + yield* discardReplacement(); + // Only a session over T3 Connect holds the old account's credential. + if (isRelayLease(lease)) { + yield* logManagedRelayAccountChange; + return "end" as const; + } } return undefined; }); @@ -682,6 +903,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( case "ConnectRequested": case "DisconnectRequested": case "BetterRouteAvailable": + case "ReplacementReady": return undefined; } }; @@ -693,6 +915,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const awaitSilentSocket = Effect.fn("EnvironmentSupervisor.awaitSilentSocket")(function* ( lease: ConnectedLease, routeKind: ConnectionRouteKind | undefined, + onHostAnswered: Effect.Effect, ) { const startedAt = yield* Clock.monotonicTimeNanos; yield* Effect.sleep(SILENT_SOCKET_CHECK_AFTER); @@ -712,6 +935,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (!(yield* socketSilent)) return "alive" as const; yield* Effect.sleep(SILENT_SOCKET_RECHECK_EVERY); } + if (!(yield* socketSilent)) return "alive" as const; + // The host takes fresh requests, so a replacement can connect while the + // socket keeps its remaining window to prove itself alive. + yield* onHostAnswered; if (!answersQuickly(routeKind)) { const silentFor = (yield* Clock.monotonicTimeNanos) - startedAt; const remaining = Duration.toNanosUnsafe(REMOTE_SILENT_SOCKET_AFTER) - silentFor; @@ -720,11 +947,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( 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 - // loopback server). - const monitorConnectedLease = Effect.fnUntraced(function* (lease: ConnectedLease) { + // Returning to the app, a retry, or an offline report (often wrong, as for a + // loopback server) probes the live session instead of tearing it down. Ends + // with true when the session should move to a replacement or a fresh attempt. + const monitorConnectedLease = Effect.fnUntraced(function* ( + lease: ConnectedLease, + generation: number, + ) { // A probe answers an explicit retry here, so the retry must not also reset // the backoff of a later, unrelated failure. const takeSignal = Queue.take(signals).pipe( @@ -742,9 +971,9 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const routeKind = yield* leaseRouteKind(lease); for (;;) { const next = yield* takeSignal; - const end = yield* connectedLeaseEnd(next, lease); + const end = yield* connectedLeaseEnd(next, lease, generation); if (end !== undefined) { - return end === "reset"; + return end === "replace"; } if ( (next._tag === "NetworkChanged" && next.network !== "offline") || @@ -764,9 +993,32 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( let recheck = false; let longResume = isLongResume(next); const startSilentSocketCheck = Effect.gen(function* () { - silentSocket = yield* Effect.forkChild(awaitSilentSocket(lease, routeKind)); + silentSocket = yield* Effect.forkChild( + awaitSilentSocket(lease, routeKind, startReplacement(generation, Option.none())), + ); recheck = false; }); + // A long suspension or a changed network path restarts a replacement + // that may have stalled with the app or on the old path. A path change + // can also strand the socket's own route (leaving the LAN), so it starts + // a replacement over whichever route answers now. + const startProbeReplacement = (signal: SupervisorSignal) => { + const pathChanged = + signal._tag === "Wakeup" && + signal.reason === "network-changed" && + routeKind !== "loopback"; + if (!pathChanged && !isLongResume(signal)) return Effect.void; + return discardReplacement().pipe( + Effect.flatMap((taken) => + Option.isSome(taken) || pathChanged + ? startReplacement( + generation, + Option.flatMap(taken, (replacement) => replacement.upgradeTo), + ) + : Effect.void, + ), + ); + }; // 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) { @@ -777,6 +1029,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } yield* startSilentSocketCheck; }); + yield* startProbeReplacement(next); yield* armSilentSocketCheck(next); const stopProbe = Effect.suspend(() => Effect.all( @@ -805,6 +1058,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ]); if (probeEvent._tag === "SocketAlive") { silentSocket = undefined; + yield* discardSuspectReplacement; if (recheck) yield* startSilentSocketCheck; continue; } @@ -828,17 +1082,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (silentSocket) yield* Fiber.interrupt(silentSocket); if (Exit.isSuccess(probeEvent.exit)) { yield* Ref.set(probeUnanswered, false); + yield* discardSuspectReplacement; } yield* probeEvent.exit; break; } - const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease); + const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease, generation); if (endDuringProbe !== undefined) { yield* stopProbe; - return endDuringProbe === "reset"; + return endDuringProbe === "replace"; } const signalTimeout = probeTimeoutFor(probeEvent.signal, routeKind); if (signalTimeout !== undefined) { + yield* startProbeReplacement(probeEvent.signal); yield* armSilentSocketCheck(probeEvent.signal); longResume ||= isLongResume(probeEvent.signal); const signalDeadline = @@ -849,26 +1105,56 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } }); + // Moving straight to a connected replacement keeps the session from reading + // as disconnected between the two. + const endAttempt = Effect.fn("EnvironmentSupervisor.endAttempt")(function* ( + exit: Exit.Exit, + ) { + const ready = + Exit.isSuccess(exit) && exit.value._tag === "Interrupted" + ? exit.value.handoff?.ready + : undefined; + // A disconnect that raced the handoff must not flash the new session. + if (ready !== undefined && Option.isSome(ready) && (yield* Ref.get(intent)).desired) { + return yield* setLease(ready.value.lease); + } + yield* clearLease; + }); + const runAttempt = Effect.fnUntraced(function* ( attempt: number, generation: number, lastFailure: ConnectionAttemptError | null, pendingRetry: Option.Option, ignoreOffline: boolean, + handoff: Option.Option, ) { - const switchingTo = yield* Ref.get(preferredRouteId); - yield* SubscriptionRef.set(prepared, Option.none()); - const synchronizedRoutes = yield* Ref.make(0); - const establishment = yield* Effect.raceAllFirst([ - exitUnlessInterrupted( - establishTracedConnection( - attempt, - generation, - lastFailure, - pendingRetry, - synchronizedRoutes, + let establish: Effect.Effect; + let synchronizedRoutes: Ref.Ref; + let startedAt = yield* Clock.monotonicTimeNanos; + const ready = Option.flatMap(handoff, (value) => value.ready); + if (Option.isSome(handoff)) { + const { replacement } = handoff.value; + yield* Effect.addFinalizer(() => closeReplacement(replacement)); + synchronizedRoutes = replacement.synchronizedRoutes; + startedAt = replacement.startedAt; + if (Option.isNone(ready)) yield* replacement.reveal; + establish = Fiber.join(replacement.fiber); + } else { + yield* SubscriptionRef.set(prepared, Option.none()); + synchronizedRoutes = yield* Ref.make(0); + establish = establishTracedConnection( + attempt, + generation, + pendingRetry, + Option.none(), + countingSynchronized(synchronizedRoutes, (progress) => + reportProgress(attempt, generation, lastFailure, progress), ), - ).pipe( + ); + } + const raceEstablishment = Effect.raceAllFirst([ + exitUnlessInterrupted(establish).pipe( Effect.map((exit): EstablishmentEvent => ({ _tag: "Completed", exit, @@ -880,18 +1166,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( resetRetry, })), ), - // Each route may use the full setup time before the next is tried, and - // each route that reaches synchronizing adds its own snapshot budget, so - // a route that fails late does not eat the next route's time. Effect.gen(function* () { - const routeCount = connectionRoutes(yield* Ref.get(currentEntry)).length; - const startedAt = yield* Clock.monotonicTimeNanos; - const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routeCount)); while (true) { - const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); - const deadline = - startedAt + setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout); - const remaining = deadline - (yield* Clock.monotonicTimeNanos); + const remaining = + (yield* setupDeadline(startedAt, synchronizedRoutes)) - + (yield* Clock.monotonicTimeNanos); if (remaining <= 0n) { return { _tag: "TimedOut" } as const satisfies EstablishmentEvent; } @@ -899,6 +1178,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } }), ]); + // A connected replacement already holds the session, so a queued signal + // must not interrupt adopting it. + const establishment: EstablishmentEvent = Option.isSome(ready) + ? { _tag: "Completed", exit: Exit.succeed(ready.value) } + : yield* raceEstablishment; if (establishment._tag === "Interrupted") { return { @@ -942,16 +1226,6 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } const active = establishment.exit.value; - if (Option.isSome(switchingTo)) { - const landed = connectionRouteId(active.lease.prepared.target); - if (landed !== switchingTo.value) { - const until = - (yield* Clock.monotonicTimeNanos) + BigInt(BETTER_ROUTE_COOLDOWN_MS) * 1_000_000n; - yield* Ref.update(routeCooldowns, (current) => - new Map(current).set(switchingTo.value, until), - ); - } - } const currentIntent = yield* Ref.get(intent); if (!currentIntent.desired || (currentIntent.network === "offline" && !ignoreOffline)) { return { @@ -970,8 +1244,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ), }; const connectedAt = yield* Clock.currentTimeMillis; - yield* SubscriptionRef.set(prepared, Option.some(connectedLease.prepared)); - yield* SubscriptionRef.set(session, Option.some(connectedLease.session)); + yield* setLease(connectedLease); if (options?.learnRoutes !== undefined) { yield* learnRoutesFrom(options.learnRoutes, connectedLease).pipe(Effect.forkScoped); } @@ -993,7 +1266,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( attemptSpan: active.attemptSpan, })), ), - monitorConnectedLease(connectedLease).pipe( + monitorConnectedLease(connectedLease, generation).pipe( Effect.mapError((error): TracedAttemptFailure => ({ error, attemptSpan: active.attemptSpan, @@ -1001,20 +1274,27 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ), ).pipe(exitUnlessInterrupted); const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt; + const stable = connectedForMs >= BACKOFF_RESET_AFTER_MS; if (Exit.isSuccess(connectedExit)) { + if (!connectedExit.value) { + yield* discardReplacement(); + return { + _tag: "Interrupted", + established: true, + stable, + resetRetry: false, + } satisfies AttemptOutcome; + } + const next = Option.getOrUndefined(yield* takeHandoff); return { _tag: "Interrupted", established: true, - stable: connectedForMs >= BACKOFF_RESET_AFTER_MS, - resetRetry: connectedExit.value, + stable, + resetRetry: true, + ...(next === undefined ? {} : { handoff: next }), } satisfies AttemptOutcome; } - const outcome = failureFromExit( - target, - connectedExit, - true, - connectedForMs >= BACKOFF_RESET_AFTER_MS, - ); + const outcome = failureFromExit(target, connectedExit, true, stable); if (outcome._tag === "Failure") { yield* Effect.void.pipe( Effect.withSpan("EnvironmentSupervisor.connectionLost", { @@ -1030,8 +1310,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }), ); } + // A replacement already opening becomes the reconnect attempt. + const next = yield* takeHandoff; + if (Option.isSome(next)) { + return { + _tag: "Interrupted", + established: true, + stable, + resetRetry: true, + handoff: next.value, + } satisfies AttemptOutcome; + } return outcome; - }, Effect.ensuring(clearLease)); + }, Effect.onExit(endAttempt)); const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) { // @effect-diagnostics-next-line raceFirstWithSleepToTimeout:off - the sleep is the retry delay (false), not a timeout around the signal loop @@ -1049,6 +1340,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( case "NetworkChanged": return false; case "BetterRouteAvailable": + case "ReplacementReady": break; } } @@ -1056,12 +1348,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ); }); - const waitForSignal = Queue.take(signals).pipe( - Effect.repeat({ while: (next) => next._tag === "BetterRouteAvailable" }), - Effect.map( - (next) => next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason), - ), - ); + const waitForSignal = Effect.gen(function* () { + for (;;) { + const next = yield* Queue.take(signals); + if (next._tag === "BetterRouteAvailable" || next._tag === "ReplacementReady") continue; + return next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason); + } + }); const run = Effect.fnUntraced(function* () { let failureCount = 0; @@ -1073,6 +1366,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( pendingRetry = Option.none(); }; let replacing = false; + let handoff = Option.none(); for (;;) { if (yield* Ref.getAndSet(resetRetryState, false)) { @@ -1082,6 +1376,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } const currentIntent = yield* Ref.get(intent); if (!currentIntent.desired) { + if (Option.isSome(handoff)) yield* closeReplacement(handoff.value.replacement); + handoff = Option.none(); resetRetryLadder(); latestFailure = null; yield* clearLease; @@ -1102,9 +1398,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const attempt = failureCount + 1; const nextGeneration = generation + 1; const outcome: AttemptOutcome = yield* Effect.scoped( - runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, replacing), + runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, replacing, handoff), ); replacing = false; + handoff = Option.none(); const failedProbe = yield* Ref.getAndSet(probeUnanswered, false); if (outcome.established) { generation = nextGeneration; @@ -1121,6 +1418,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (outcome.resetRetry) { resetRetryLadder(); replacing = true; + handoff = Option.fromUndefinedOr(outcome.handoff); } continue; } From 78c115e1f5bea41f9baaf009a6c9027471bf13d3 Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Sat, 7 Nov 2026 11:46:07 -0800 Subject: [PATCH 2/4] perf(mobile): cold starts feel faster and reconnect more quietly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sticky routes, hedged connects, quieter status, LAN skip, stale deferral, DPoP prefetch, early ticket, ready-on-WS-open, wake renew, and relay endpoint reuse — stacked on seamless resume from #30835. Co-authored-by: Cursor --- .../src/app/DesktopConnectionCatalogStore.ts | 1 + apps/mobile/src/connection/platform.ts | 18 ++ apps/mobile/src/connection/storage.ts | 17 +- apps/web/src/connection/platform.ts | 2 + apps/web/src/connection/storage.test.ts | 1 + apps/web/src/connection/storage.ts | 16 ++ docs/internals/connection-runtime.md | 13 +- .../src/authorization/layer.test.ts | 90 ++++++-- .../src/authorization/service.ts | 156 ++++++++++--- .../src/connection/connectivity.ts | 5 + .../client-runtime/src/connection/driver.ts | 210 +++++++++++++++-- .../client-runtime/src/connection/model.ts | 2 + .../src/connection/onboarding.test.ts | 22 +- .../src/connection/registry.test.ts | 212 +++++++++++++++++- .../client-runtime/src/connection/registry.ts | 180 +++++++++++++-- .../src/connection/routes.test.ts | 35 +++ .../client-runtime/src/connection/routes.ts | 40 ++++ .../src/connection/supervisor.test.ts | 127 ++++++++++- .../src/connection/supervisor.ts | 16 +- .../src/platform/persistence.ts | 10 + .../src/platform/storageDocument.test.ts | 25 +++ .../src/platform/storageDocument.ts | 33 +++ .../src/relay/discovery.test.ts | 2 + .../client-runtime/src/rpc/session.test.ts | 20 +- packages/client-runtime/src/rpc/session.ts | 11 +- 25 files changed, 1151 insertions(+), 113 deletions(-) diff --git a/apps/desktop/src/app/DesktopConnectionCatalogStore.ts b/apps/desktop/src/app/DesktopConnectionCatalogStore.ts index 3240df3636e38..663bcb71b48aa 100644 --- a/apps/desktop/src/app/DesktopConnectionCatalogStore.ts +++ b/apps/desktop/src/app/DesktopConnectionCatalogStore.ts @@ -372,6 +372,7 @@ const migrateSavedEnvironmentRecords = Effect.fn( credentials, remoteDpopTokens: [], disabledEnvironmentIds: [], + lastConnected: [], }; }); diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index 407016ab484d9..5cac44ddf6a33 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -41,6 +41,15 @@ function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "on return "unknown"; } +function privateNetworkLikely(state: Network.NetworkState): boolean { + if (state.isConnected !== true) return false; + return ( + state.type === Network.NetworkStateType.WIFI || + state.type === Network.NetworkStateType.ETHERNET || + state.type === Network.NetworkStateType.UNKNOWN + ); +} + const layerConnectivity = Connectivity.layer({ status: Effect.tryPromise({ try: () => Network.getNetworkStateAsync(), @@ -51,6 +60,15 @@ const layerConnectivity = Connectivity.layer({ onSuccess: networkStatus, }), ), + privateNetworkLikely: Effect.tryPromise({ + try: () => Network.getNetworkStateAsync(), + catch: () => undefined, + }).pipe( + Effect.match({ + onFailure: () => true, + onSuccess: (state) => (state === undefined ? true : privateNetworkLikely(state)), + }), + ), changes: Stream.callback((queue) => Effect.acquireRelease( Effect.sync(() => { diff --git a/apps/mobile/src/connection/storage.ts b/apps/mobile/src/connection/storage.ts index 91a6ac71919da..2b800d8afee58 100644 --- a/apps/mobile/src/connection/storage.ts +++ b/apps/mobile/src/connection/storage.ts @@ -4,6 +4,7 @@ import { removeConnectionFromCatalog, setConnectionEnabledInCatalog, setRoutesInCatalog, + touchEnvironmentConnectedInCatalog, removeCatalogValue, replaceCatalogValue, Persistence, @@ -27,10 +28,12 @@ function targetPersistenceError( operation: | "list-targets" | "list-disabled-targets" + | "list-last-connected" | "register-connection" | "set-connection-routes" | "remove-connection" - | "set-connection-enabled", + | "set-connection-enabled" + | "touch-connection", error: ConnectionTransientError, ) { return new Persistence.ConnectionPersistenceError({ @@ -57,6 +60,10 @@ export const layer = Layer.effectContext( Effect.map((document) => document.disabledEnvironmentIds), Effect.mapError((error) => targetPersistenceError("list-disabled-targets", error)), ), + listLastConnected: catalog.read.pipe( + Effect.map((document) => document.lastConnected), + Effect.mapError((error) => targetPersistenceError("list-last-connected", error)), + ), }); const registrationStore = Persistence.ConnectionRegistrationStore.of({ register: Effect.fn("Persistence.ConnectionRegistrationStore.register")( @@ -86,6 +93,14 @@ export const layer = Layer.effectContext( Effect.mapError((error) => targetPersistenceError("set-connection-enabled", error)), ), ), + touchConnected: Effect.fn("Persistence.ConnectionRegistrationStore.touchConnected")( + (environmentId, atEpochMs) => + catalog + .update((document) => + touchEnvironmentConnectedInCatalog(document, environmentId, atEpochMs), + ) + .pipe(Effect.mapError((error) => targetPersistenceError("touch-connection", error))), + ), }); const profileStore = ProfileStore.make({ get: (connectionId) => diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index dcd50342f1c91..5d06a4a1fa79c 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -74,6 +74,8 @@ function currentNetworkStatus(): "unknown" | "offline" | "online" { const layerConnectivity = Connectivity.layer({ status: Effect.sync(currentNetworkStatus), + // Browsers rarely expose a reliable cellular/Wi-Fi split; keep LAN routes eligible. + privateNetworkLikely: Effect.succeed(true), changes: Stream.callback((queue) => Effect.acquireRelease( Effect.sync(() => { diff --git a/apps/web/src/connection/storage.test.ts b/apps/web/src/connection/storage.test.ts index 7afe2b1e8573f..8df789cce38ea 100644 --- a/apps/web/src/connection/storage.test.ts +++ b/apps/web/src/connection/storage.test.ts @@ -21,6 +21,7 @@ const emptyCatalog = { credentials: [], remoteDpopTokens: [], disabledEnvironmentIds: [], + lastConnected: [], } as const; const decodeCatalog = Schema.decodeUnknownSync(Schema.fromJsonString(ConnectionCatalogDocument)); const encodeCatalog = Schema.encodeSync(Schema.fromJsonString(ConnectionCatalogDocument)); diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 78b01fc6d4695..29dc38748ed9f 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -13,6 +13,7 @@ import { removeConnectionFromCatalog, setConnectionEnabledInCatalog, setRoutesInCatalog, + touchEnvironmentConnectedInCatalog, replaceCatalogValue, Persistence, makeStoredShellSnapshotEncoder, @@ -104,10 +105,12 @@ function persistenceError( operation: | "list-targets" | "list-disabled-targets" + | "list-last-connected" | "register-connection" | "set-connection-routes" | "remove-connection" | "set-connection-enabled" + | "touch-connection" | "load-shell" | "save-shell" | "load-thread" @@ -733,6 +736,10 @@ export const layer = Layer.effectContext( Effect.map((document) => document.disabledEnvironmentIds), Effect.mapError((cause) => persistenceError("list-disabled-targets", cause)), ), + listLastConnected: catalog.read.pipe( + Effect.map((document) => document.lastConnected), + Effect.mapError((cause) => persistenceError("list-last-connected", cause)), + ), }); const registrationStore = Persistence.ConnectionRegistrationStore.of({ register: Effect.fn("Persistence.ConnectionRegistrationStore.register")( @@ -761,6 +768,15 @@ export const layer = Layer.effectContext( .pipe(Effect.mapError((cause) => persistenceError("set-connection-enabled", cause))); }, ), + touchConnected: Effect.fn("Persistence.ConnectionRegistrationStore.touchConnected")( + function (environmentId, atEpochMs) { + return catalog + .update((document) => + touchEnvironmentConnectedInCatalog(document, environmentId, atEpochMs), + ) + .pipe(Effect.mapError((cause) => persistenceError("touch-connection", cause))); + }, + ), }); const profileStore = ProfileStore.make({ get: (connectionId) => diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 961bdb261a5d7..ba3b16e84f177 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -76,12 +76,13 @@ require valid credentials. ## Transport health and data freshness are separate -A socket opening is insufficient evidence that the environment is usable. The -[RPC session](../../packages/client-runtime/src/rpc/session.ts) waits for the -initial server configuration before becoming ready. Shell and thread data then -have their own synchronization state. A failed shell subscription can coexist -with a healthy connection; labeling that state "reconnecting" promises a -transport retry that will never happen. +The [RPC session](../../packages/client-runtime/src/rpc/session.ts) becomes ready +when the WebSocket opens so the UI can leave "Connecting…" without waiting on a +large config snapshot. Prefer `initialConfig` / `subscribeServerConfig` when the +snapshot itself is required. Shell and thread data then have their own +synchronization state. A failed shell subscription can coexist with a healthy +connection; labeling that state "reconnecting" promises a transport retry that +will never happen. Cached projections remain readable offline. They must neither imply a live connection nor overwrite newer live data during a reconnect. Loading and diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 9ed7b9a63e9d3..baf8f88f476e6 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -500,11 +500,8 @@ describe("RemoteEnvironmentAuthorization", () => { }); const harness = yield* makeHarness({ initialToken: expired, - responses: [ - Response.json(DESCRIPTOR), - accessToken("fresh-access-token"), - websocketTicket("fresh-ticket"), - ], + // Prior tunnel host is reused, so the descriptor fetch is skipped. + responses: [accessToken("fresh-access-token"), websocketTicket("fresh-ticket")], }); const authorized = yield* Effect.gen(function* () { @@ -520,9 +517,64 @@ describe("RemoteEnvironmentAuthorization", () => { expect.objectContaining({ accessToken: "fresh-access-token", dpopThumbprint: "thumbprint-1", + label: DESCRIPTOR.label, }), ); - expect(harness.fetch.calls).toHaveLength(3); + expect(harness.fetch.calls).toHaveLength(2); + }), + ); + + it.effect("renews early when authorizeDpopHttp asks for a wider refresh skew", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + initialToken: persistedToken({ expiresAtEpochMs: 4 * 60_000 }), + responses: [accessToken("wake-renewed-token")], + }); + + const authorized = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + return yield* remote.authorizeDpopHttp({ + expectedEnvironmentId: ENVIRONMENT_ID, + refreshSkewMs: 5 * 60_000, + }); + }).pipe(Effect.provide(harness.layer)); + + expect(authorized.httpAuthorization.accessToken).toBe("wake-renewed-token"); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); + }), + ); + + it.effect("fetches a descriptor when the relay returns a new tunnel endpoint", () => + Effect.gen(function* () { + const moved = { + ...ENDPOINT, + httpBaseUrl: "https://moved.example.test", + wsBaseUrl: "wss://moved.example.test", + }; + const harness = yield* makeHarness({ + initialToken: persistedToken({ expiresAtEpochMs: 0 }), + bootstrap: { ...BOOTSTRAP, endpoint: moved }, + responses: [ + Response.json({ ...DESCRIPTOR, label: "Moved environment" }), + accessToken("moved-access-token"), + websocketTicket("moved-ticket"), + ], + }); + + const authorized = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + return yield* remote.authorizeDpop({ + expectedEnvironmentId: ENVIRONMENT_ID, + }); + }).pipe(Effect.provide(harness.layer)); + + expect(authorized.socketUrl).toContain("wsTicket=moved-ticket"); + expect(authorized.label).toBe("Moved environment"); + expect(harness.fetch.calls.map(([url]) => String(url))).toEqual([ + "https://moved.example.test/.well-known/t3/environment", + "https://moved.example.test/oauth/token", + "https://moved.example.test/api/auth/websocket-ticket/rpc", + ]); }), ); @@ -556,7 +608,6 @@ describe("RemoteEnvironmentAuthorization", () => { Response.json(DESCRIPTOR), exchangeGrant, websocketTicket("granted-ticket"), - Response.json(DESCRIPTOR), exchangeGrant, ], }); @@ -621,12 +672,7 @@ describe("RemoteEnvironmentAuthorization", () => { ), ), initialToken: persistedToken({ expiresAtEpochMs: 3_600_000 }), - responses: [ - Response.json(DESCRIPTOR), - accessToken("renewed-1"), - Response.json(DESCRIPTOR), - accessToken("renewed-2"), - ], + responses: [accessToken("renewed-1"), accessToken("renewed-2")], beforePut: Queue.offer(puts, undefined).pipe(Effect.asVoid), }); @@ -670,7 +716,6 @@ describe("RemoteEnvironmentAuthorization", () => { initialToken: cached, responses: [ authInvalid(), - Response.json(DESCRIPTOR), accessToken("replacement-access-token"), websocketTicket("replacement-ticket"), ], @@ -690,7 +735,7 @@ describe("RemoteEnvironmentAuthorization", () => { accessToken: "replacement-access-token", }), ); - expect(harness.fetch.calls).toHaveLength(4); + expect(harness.fetch.calls).toHaveLength(3); }), ); @@ -709,7 +754,6 @@ describe("RemoteEnvironmentAuthorization", () => { initialToken: cached, responses: [ new Response("endpoint unavailable", { status: 503 }), - Response.json(DESCRIPTOR), accessToken("replacement-access-token"), websocketTicket("replacement-ticket"), ], @@ -729,7 +773,7 @@ describe("RemoteEnvironmentAuthorization", () => { accessToken: "replacement-access-token", }), ); - expect(harness.fetch.calls).toHaveLength(4); + expect(harness.fetch.calls).toHaveLength(3); }), ); @@ -760,7 +804,7 @@ describe("RemoteEnvironmentAuthorization", () => { Effect.gen(function* () { const harness = yield* makeHarness({ initialToken: persistedToken(), - responses: [Response.json(DESCRIPTOR), accessToken("replacement-token")], + responses: [accessToken("replacement-token")], }); yield* Effect.gen(function* () { const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; @@ -776,7 +820,7 @@ describe("RemoteEnvironmentAuthorization", () => { }).pipe(Effect.provide(harness.layer)); expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); - expect(harness.fetch.calls).toHaveLength(2); + expect(harness.fetch.calls).toHaveLength(1); expect((yield* Ref.get(harness.tokens)).get(ENVIRONMENT_ID)?.accessToken).toBe( "replacement-token", ); @@ -792,7 +836,7 @@ describe("RemoteEnvironmentAuthorization", () => { beforeBootstrap: Deferred.succeed(started, undefined).pipe( Effect.andThen(Deferred.await(release)), ), - responses: [Response.json(DESCRIPTOR), accessToken("shared-token")], + responses: [accessToken("shared-token")], }); yield* Effect.gen(function* () { const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; @@ -813,7 +857,7 @@ describe("RemoteEnvironmentAuthorization", () => { }).pipe(Effect.provide(harness.layer)); expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); - expect(harness.fetch.calls).toHaveLength(2); + expect(harness.fetch.calls).toHaveLength(1); }), ); @@ -1025,7 +1069,8 @@ describe("RemoteEnvironmentAuthorization", () => { Effect.gen(function* () { const harness = yield* makeHarness({ initialToken: persistedToken(), - responses: [Response.json(DESCRIPTOR), accessToken("new-session-token")], + // Same account can reuse the prior tunnel host; only the oauth exchange runs. + responses: [accessToken("new-session-token")], }); yield* Effect.gen(function* () { const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; @@ -1036,6 +1081,7 @@ describe("RemoteEnvironmentAuthorization", () => { expect(signedIn.httpAuthorization.accessToken).toBe("new-session-token"); }).pipe(Effect.provide(harness.layer)); expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); + expect(harness.fetch.calls).toHaveLength(1); }), ); diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 9b4d9c2f1f892..270a5ade5b53d 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -43,6 +43,9 @@ import { type PreparedHttpAuthorization, } from "../connection/model.ts"; +/** Reuse a prior tunnel host's label when connect returns the same endpoint. */ +const RELAY_ENDPOINT_HINT_TTL_MS = 15 * 60_000; + export interface AuthorizedRemoteEnvironment { readonly environmentId: EnvironmentId; readonly label: string; @@ -81,6 +84,8 @@ export class RemoteEnvironmentAuthorization extends Context.Service< readonly authorizeDpopHttp: (input: { readonly expectedEnvironmentId: EnvironmentId; readonly rejectedAccessToken?: string; + /** Treat cached tokens as stale this far before expiry. Defaults to the live-socket skew. */ + readonly refreshSkewMs?: number; }) => Effect.Effect; /** * Renews a live socket's token before it expires, so the server moves the @@ -134,6 +139,30 @@ export const make = Effect.gen(function* () { readonly token: Effect.Effect; } >(); + const lastRelayEndpoints = new Map< + EnvironmentId, + { + readonly accountId: string; + readonly label: string; + readonly endpoint: TokenStore.RemoteDpopAccessToken["endpoint"]; + readonly atEpochMs: number; + } + >(); + + const rememberRelayEndpoint = (token: TokenStore.RemoteDpopAccessToken, atEpochMs: number) => { + if (token.accountId === undefined) return; + lastRelayEndpoints.set(token.environmentId, { + accountId: token.accountId, + label: token.label, + endpoint: token.endpoint, + atEpochMs, + }); + }; + + const sameRelayEndpoint = ( + left: TokenStore.RemoteDpopAccessToken["endpoint"], + right: TokenStore.RemoteDpopAccessToken["endpoint"], + ) => left.httpBaseUrl === right.httpBaseUrl && left.wsBaseUrl === right.wsBaseUrl; /** * A saved LAN address can belong to another machine on a different network. @@ -261,11 +290,12 @@ export const make = Effect.gen(function* () { identity: ClientCapabilities.CloudSessionIdentity, ) { yield* assertSession(identity); - const clerkToken = yield* cloudSession.clerkToken.pipe( - Effect.withSpan("relay.connection.cloudSessionToken.resolve"), - ); - const deviceId = yield* deviceIdentity.deviceId.pipe( - Effect.withSpan("relay.connection.deviceIdentity.resolve"), + const [clerkToken, deviceId] = yield* Effect.all( + [ + cloudSession.clerkToken.pipe(Effect.withSpan("relay.connection.cloudSessionToken.resolve")), + deviceIdentity.deviceId.pipe(Effect.withSpan("relay.connection.deviceIdentity.resolve")), + ], + { concurrency: 2 }, ); yield* assertSession(identity); const connected = yield* relay @@ -286,37 +316,95 @@ export const make = Effect.gen(function* () { return connected; }); + const createBootstrapProof = Effect.fnUntraced(function* (httpBaseUrl: string) { + return yield* signer + .createProof({ + method: "POST", + url: environmentEndpointUrl(httpBaseUrl, "/oauth/token"), + }) + .pipe( + Effect.mapError( + () => + new ConnectionBlockedError({ + reason: "configuration", + detail: "Could not create the environment authorization proof.", + }), + ), + ); + }); + const exchangeDpopToken = Effect.fn("clientRuntime.connection.remote.exchangeDpopToken")( function* ( environmentId: EnvironmentId, thumbprint: string, identity: ClientCapabilities.CloudSessionIdentity, ) { + const now = yield* Clock.currentTimeMillis; + const priorToken = yield* tokenStore.get(environmentId); + const storedHint = lastRelayEndpoints.get(environmentId); + // A still-stored token is authoritative for its endpoint. The TTL only + // bounds the in-memory hint after the token row was cleared. + const reusableHint = + Option.isSome(priorToken) && + priorToken.value.accountId === identity.accountId && + priorToken.value.dpopThumbprint === thumbprint + ? { + accountId: identity.accountId, + label: priorToken.value.label, + endpoint: priorToken.value.endpoint, + atEpochMs: now, + } + : storedHint !== undefined && + storedHint.accountId === identity.accountId && + now - storedHint.atEpochMs < RELAY_ENDPOINT_HINT_TTL_MS + ? storedHint + : undefined; + + // Mint the DPoP proof against the last tunnel host while the relay assigns + // a bootstrap credential; most renewals land on the same endpoint. + const speculativeProof = + reusableHint === undefined + ? undefined + : yield* Effect.forkChild(createBootstrapProof(reusableHint.endpoint.httpBaseUrl)); const bootstrap = yield* obtainBootstrap(environmentId, identity); - const descriptor = yield* fetchDescriptor(bootstrap.endpoint.httpBaseUrl, "relay").pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - Effect.withSpan("environment.authorization.descriptor"), - ); - if (descriptor.environmentId !== environmentId) { - return yield* environmentMismatchError({ - expected: environmentId, - actual: descriptor.environmentId, - }); + const endpointReused = + reusableHint !== undefined && sameRelayEndpoint(reusableHint.endpoint, bootstrap.endpoint); + yield* Effect.annotateCurrentSpan({ + "connection.relay_endpoint_cache": endpointReused ? "hit" : "miss", + }); + + if (speculativeProof !== undefined && !endpointReused) { + yield* Fiber.interrupt(speculativeProof).pipe(Effect.ignore); } - const bootstrapProof = yield* signer - .createProof({ - method: "POST", - url: environmentEndpointUrl(bootstrap.endpoint.httpBaseUrl, "/oauth/token"), - }) - .pipe( - Effect.mapError( - () => - new ConnectionBlockedError({ - reason: "configuration", - detail: "Could not create the environment authorization proof.", - }), - ), - ); + + const labelAndProof: { readonly label: string; readonly proof: string } = endpointReused + ? { + label: reusableHint.label, + proof: + speculativeProof !== undefined + ? yield* Fiber.join(speculativeProof) + : yield* createBootstrapProof(bootstrap.endpoint.httpBaseUrl), + } + : yield* Effect.all( + { + label: fetchDescriptor(bootstrap.endpoint.httpBaseUrl, "relay").pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.withSpan("environment.authorization.descriptor"), + Effect.flatMap((descriptor) => + descriptor.environmentId === environmentId + ? Effect.succeed(descriptor.label) + : environmentMismatchError({ + expected: environmentId, + actual: descriptor.environmentId, + }), + ), + ), + proof: createBootstrapProof(bootstrap.endpoint.httpBaseUrl), + }, + { concurrency: 2 }, + ); + const { label, proof: bootstrapProof } = labelAndProof; + yield* assertSession(identity); const access = yield* exchangeRemoteDpopAccessToken({ httpBaseUrl: bootstrap.endpoint.httpBaseUrl, @@ -329,15 +417,17 @@ export const make = Effect.gen(function* () { Effect.withSpan("environment.authorization.accessToken.exchange"), ); const issuedAt = yield* Clock.currentTimeMillis; - return new TokenStore.RemoteDpopAccessToken({ - environmentId: descriptor.environmentId, + const token = new TokenStore.RemoteDpopAccessToken({ + environmentId, accountId: identity.accountId, - label: descriptor.label, + label, endpoint: bootstrap.endpoint, accessToken: access.access_token, expiresAtEpochMs: issuedAt + access.expires_in * 1_000, dpopThumbprint: thumbprint, }); + rememberRelayEndpoint(token, issuedAt); + return token; }, ); @@ -359,6 +449,7 @@ export const make = Effect.gen(function* () { ), Effect.withSpan("environment.authorization.dpopKey.resolve"), ); + const refreshSkewMs = input.refreshSkewMs ?? DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS; const selected = yield* tokenLock.withPermits(1)( Effect.gen(function* () { yield* assertSession(identity); @@ -372,7 +463,7 @@ export const make = Effect.gen(function* () { cached.value.environmentId === input.expectedEnvironmentId && cached.value.accountId === identity.accountId && cached.value.dpopThumbprint === thumbprint && - cached.value.expiresAtEpochMs > now + DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS && + cached.value.expiresAtEpochMs > now + refreshSkewMs && cached.value.accessToken !== input.rejectedAccessToken && (owner?.accessToken !== cached.value.accessToken || owner.identity === identity) ) { @@ -380,6 +471,7 @@ export const make = Effect.gen(function* () { accessToken: cached.value.accessToken, identity, }); + rememberRelayEndpoint(cached.value, now); return { token: Effect.succeed(cached.value), fromCache: true }; } if (input.rejectedAccessToken !== undefined) { diff --git a/packages/client-runtime/src/connection/connectivity.ts b/packages/client-runtime/src/connection/connectivity.ts index 9bf9dbdb24090..99bb0a170ded5 100644 --- a/packages/client-runtime/src/connection/connectivity.ts +++ b/packages/client-runtime/src/connection/connectivity.ts @@ -10,6 +10,11 @@ export class Connectivity extends Context.Service< { readonly status: Effect.Effect; readonly changes: Stream.Stream; + /** + * False on cellular and similar paths where RFC1918 LAN addresses cannot + * be reached. Defaults to true when the platform cannot tell. + */ + readonly privateNetworkLikely: Effect.Effect; } >()("@t3tools/client-runtime/connection/connectivity") {} diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index d644e62d41025..b0dd5dfc7c73e 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -1,13 +1,19 @@ +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; 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"; +import * as Connectivity from "./connectivity.ts"; import type { ConnectionAttemptError, ConnectionAttemptStage, @@ -16,11 +22,15 @@ import type { import { ConnectionTransientError } from "./model.ts"; import * as ConnectionResolver from "./resolver.ts"; import { + connectionRouteId, + connectionRouteKind, connectionRoutes, routeEntry, routeHttpBaseUrl, + routeUsesDpop, + routesForNetwork, + routesIncludeLan, sendsTokenInClear, - usableRoutes, } from "./routes.ts"; import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; import * as RpcSession from "../rpc/session.ts"; @@ -49,6 +59,18 @@ const ROUTE_CHECK_TIMEOUT_MS = 2_500; // wait the full probe timeout for a LAN the device has already left. A route // that misses it connects over relay, then upgrades over a second connection. const ROUTE_CHECK_HEAD_START_MS = 500; +// Give the preferred ready route a short solo window before parallel backups +// start, so a fast LAN does not also open a relay session. +const ROUTE_CONNECT_HEDGE_MS = 250; + +export interface ConnectOverRoutesOptions { + readonly privateNetworkLikely?: boolean; + /** + * When false, routes are tried one at a time. Better-route upgrades need this + * so a hanging preferred path is not "won" by a parallel backup of the live route. + */ + readonly hedge?: boolean; +} export class ConnectionDriver extends Context.Service< ConnectionDriver, @@ -56,6 +78,7 @@ export class ConnectionDriver extends Context.Service< readonly connect: ( entry: ConnectionCatalogEntry, reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect, + options?: ConnectOverRoutesOptions, ) => Effect.Effect; readonly preflight: ( entry: ConnectionCatalogEntry, @@ -72,13 +95,20 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") connectRoute: ( route: ConnectionRoute, ) => Effect.Effect, + options: ConnectOverRoutesOptions = {}, ) { - const routes = usableRoutes(entry); - const checks = - routes.length === 1 - ? [] - : yield* Effect.forEach(routes, (route) => Effect.forkChild(checkRoute(route))); - const headStart = yield* Effect.forkChild(Effect.sleep(ROUTE_CHECK_HEAD_START_MS)); + const privateNetworkLikely = options.privateNetworkLikely ?? true; + const hedge = options.hedge ?? true; + const routes = routesForNetwork(entry, privateNetworkLikely); + // Reachability checks exist to avoid opening a dead LAN before relay. With no + // LAN in the race, start connecting immediately and let the hedge pick a winner. + const classifyFirst = routes.length > 1 && routesIncludeLan(routes); + const checks = classifyFirst + ? yield* Effect.forEach(routes, (route) => Effect.forkChild(checkRoute(route))) + : []; + const headStart = classifyFirst + ? yield* Effect.forkChild(Effect.sleep(ROUTE_CHECK_HEAD_START_MS)) + : undefined; const attemptScope = yield* Scope.Scope; let transient: ConnectionAttemptError | undefined; let blocked: ConnectionAttemptError | undefined; @@ -93,13 +123,106 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") if (result._tag === "Failure") { if (result.failure._tag === "ConnectionTransientError") transient ??= result.failure; else blocked ??= result.failure; + return { result, routeScope } as const; + } + return { result, routeScope } as const; + }); + + // First success wins. Later successes close their scopes so a hedged backup + // cannot leave a second live session behind. A finished failure starts the + // next route immediately; only a still-running preferred attempt waits out + // the hedge before opening a parallel backup. + const raceAttempts = Effect.fnUntraced(function* (candidates: ReadonlyArray) { + if (candidates.length === 0) { + return undefined as + | Result.Result + | undefined; + } + if (candidates.length === 1) { + const { result } = yield* attempt(candidates[0]!); + return result; + } + const decided = + yield* Deferred.make>(); + const remaining = yield* Ref.make(candidates.length); + const fibers: Array> = []; + let previous = Option.none>(); + + for (const [index, route] of candidates.entries()) { + if (index > 0 && Option.isSome(previous)) { + const gate = hedge + ? yield* Effect.raceFirst( + Deferred.await(decided).pipe(Effect.as("decided" as const)), + Fiber.join(previous.value).pipe(Effect.as("finished" as const)), + ).pipe( + Effect.timeoutOrElse({ + duration: ROUTE_CONNECT_HEDGE_MS, + orElse: () => Effect.succeed("hedge" as const), + }), + ) + : yield* Effect.raceFirst( + Deferred.await(decided).pipe(Effect.as("decided" as const)), + Fiber.join(previous.value).pipe(Effect.as("finished" as const)), + ); + if (gate === "decided") break; + } + // Only connectRoute is interruptible. Forking the scope, claiming the + // winner, and closing a loser must stay masked — restore(attempt) would + // leave a gap after onExit sees success and before the claim runs. + const fiber = yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const routeScope = yield* Scope.fork(attemptScope); + // Only the connect attempt is interruptible. onExit/result/claim stay + // masked so a success cannot be interrupted before the winner is chosen. + const result = yield* restore(connectRoute(route).pipe(Scope.provide(routeScope))).pipe( + Effect.onExit((exit) => + Exit.isSuccess(exit) ? Effect.void : Scope.close(routeScope, exit), + ), + Effect.result, + ); + if (result._tag === "Failure") { + if (result.failure._tag === "ConnectionTransientError") transient ??= result.failure; + else blocked ??= result.failure; + } + if (result._tag === "Success" || result.failure.reason === "unsupported") { + // Deferred.succeed's boolean is the claim; discarding it leaves a + // losing hedged session open for the life of the winner. + const claimed = yield* Deferred.succeed(decided, result).pipe( + Effect.orElseSucceed(() => false), + ); + if (!claimed) yield* Scope.close(routeScope, Exit.void); + return; + } + const count = yield* Ref.updateAndGet(remaining, (n) => n - 1); + if (count === 0) yield* Deferred.succeed(decided, result).pipe(Effect.ignore); + }), + ).pipe(Effect.forkChild); + fibers.push(fiber); + previous = Option.some(fiber); } + + const result = yield* Deferred.await(decided); + yield* Effect.forEach(fibers, (fiber) => Fiber.interrupt(fiber), { + concurrency: "unbounded", + discard: true, + }); return result; }); + + const settle = ( + result: Result.Result | undefined, + ) => { + if (result === undefined) return Effect.undefined; + if (result._tag === "Success") return Effect.succeed(result.success); + if (result.failure.reason === "unsupported") return Effect.fail(result.failure); + return Effect.undefined; + }; + const silent: Array<{ readonly route: ConnectionRoute; readonly index: number }> = []; + const ready: Array = []; for (const [index, route] of routes.entries()) { const check = - checks.length === 0 + headStart === undefined || checks.length === 0 ? "unchecked" : yield* Effect.raceFirst( Fiber.join(checks[index]!), @@ -109,25 +232,24 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") silent.push({ route, index }); continue; } - const result = yield* attempt(route); - if (result._tag === "Success") return result.success; - // An incompatible server is the same server on every route. - if (result.failure.reason === "unsupported") return yield* result.failure; + ready.push(route); } + + const first = yield* settle(yield* raceAttempts(ready)); + if (first !== undefined) return first; + // A slow route that answered after the head start goes before one that is // still dead, which can hold an attempt open far longer than its probe. const late = yield* Effect.forEach(silent, ({ index }) => Fiber.join(checks[index]!), { concurrency: "unbounded", }); const retry = [ - ...silent.filter((_, position) => late[position] !== "silent"), - ...silent.filter((_, position) => late[position] === "silent"), + ...silent.filter((_, position) => late[position] !== "silent").map(({ route }) => route), + ...silent.filter((_, position) => late[position] === "silent").map(({ route }) => route), ]; - for (const { route } of retry) { - const result = yield* attempt(route); - if (result._tag === "Success") return result.success; - if (result.failure.reason === "unsupported") return yield* result.failure; - } + const second = yield* settle(yield* raceAttempts(retry)); + if (second !== undefined) return second; + return yield* ( transient ?? blocked ?? @@ -144,6 +266,7 @@ export const make = Effect.gen(function* () { const sessions = yield* RpcSession.RpcSessionFactory; const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; const httpClient = yield* HttpClient.HttpClient; + const connectivity = yield* Connectivity.Connectivity; // The descriptor is public, so this sends no credential to whatever // answers at a saved LAN address on a different network. @@ -181,23 +304,69 @@ export const make = Effect.gen(function* () { const connect = Effect.fn("ConnectionDriver.connect")(function* ( entry: ConnectionCatalogEntry, reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect, + options: ConnectOverRoutesOptions = {}, ) { const target = entry.target; + const startedAt = yield* Clock.currentTimeMillis; yield* Effect.annotateCurrentSpan({ "connection.environment.id": target.environmentId, "connection.target.kind": target._tag, "connection.route.count": connectionRoutes(entry).length, + "connection.stage": "preparing", }); yield* reportProgress({ stage: "preparing" }); + const privateNetworkLikely = + options.privateNetworkLikely ?? (yield* connectivity.privateNetworkLikely); + const hedge = options.hedge ?? true; + const routes = routesForNetwork(entry, privateNetworkLikely); + const preferred = routes[0]; + // Mint the DPoP ticket while route checks run so T3 Connect is not strictly + // serialized behind the LAN head-start window. + const preferredPrepare = + preferred !== undefined && routeUsesDpop(preferred) + ? yield* Effect.forkChild( + resolver.prepare(routeEntry(entry, preferred)).pipe(Effect.result), + ) + : undefined; + const preferredRouteId = + preferred === undefined ? undefined : connectionRouteId(preferred.target); return yield* connectOverRoutes( entry, (route) => checkRoute(entry, route), Effect.fnUntraced(function* (route) { - const prepared = yield* resolver.prepare(routeEntry(entry, route)); + const prepared = + preferredPrepare !== undefined && + preferredRouteId !== undefined && + connectionRouteId(route.target) === preferredRouteId + ? yield* Fiber.join(preferredPrepare).pipe( + Effect.flatMap((result) => + result._tag === "Success" + ? Effect.succeed(result.success) + : Effect.fail(result.failure), + ), + ) + : yield* resolver.prepare(routeEntry(entry, route)); + const preparedAt = yield* Clock.currentTimeMillis; + yield* Effect.annotateCurrentSpan({ + "connection.stage": "opening", + "connection.prepare_ms": preparedAt - startedAt, + "connection.route.kind": connectionRouteKind(route), + }); yield* reportProgress({ stage: "opening", prepared }); const session = yield* sessions.connect(prepared); + const openingAt = yield* Clock.currentTimeMillis; + yield* Effect.annotateCurrentSpan({ + "connection.stage": "synchronizing", + "connection.open_ms": openingAt - preparedAt, + }); yield* reportProgress({ stage: "synchronizing", prepared }); yield* session.ready; + const readyAt = yield* Clock.currentTimeMillis; + yield* Effect.annotateCurrentSpan({ + "connection.stage": "ready", + "connection.ready_ms": readyAt - openingAt, + "connection.total_ms": readyAt - startedAt, + }); // The server closes a socket when its session expires, unless the same key renewed it. if (prepared.httpAuthorization?._tag === "Dpop") { yield* remote @@ -214,6 +383,7 @@ export const make = Effect.gen(function* () { } return { prepared, session } satisfies EnvironmentConnectionLease; }), + { privateNetworkLikely, hedge }, ); }); diff --git a/packages/client-runtime/src/connection/model.ts b/packages/client-runtime/src/connection/model.ts index 73675d431aa7a..bca85f8f7bc57 100644 --- a/packages/client-runtime/src/connection/model.ts +++ b/packages/client-runtime/src/connection/model.ts @@ -135,6 +135,8 @@ export class ConnectionBlockedError extends Schema.TaggedError Effect.void, setEnabled: () => Effect.void, + touchConnected: () => Effect.void, }), Layer.mock(Persistence.EnvironmentCacheStore)({}), Layer.mock(ConnectionProfileStore.ConnectionProfileStore)({}), @@ -183,13 +186,30 @@ function layerPairingOnboarding( Layer.mock(ClientCapabilities.SshEnvironmentGateway)({}), Layer.succeed( Connectivity.Connectivity, - Connectivity.Connectivity.of({ status: Effect.succeed("online"), changes: Stream.never }), + Connectivity.Connectivity.of({ + status: Effect.succeed("online"), + changes: Stream.never, + privateNetworkLikely: Effect.succeed(true), + }), ), Layer.mock(ConnectionDriver.ConnectionDriver)({ connect: () => Effect.never }), Layer.succeed( ConnectionWakeups.ConnectionWakeups, ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.never }), ), + Layer.mock(RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization)({ + authorizeDpopHttp: (input) => + Effect.succeed({ + environmentId: input.expectedEnvironmentId, + label: "Prefetched", + httpBaseUrl: "https://relay.example.test", + httpAuthorization: { + _tag: "Dpop" as const, + accessToken: "prefetch-token", + expiresAtEpochMs: Number.MAX_SAFE_INTEGER, + }, + }), + }), ); return Onboarding.layer.pipe( Layer.provideMerge(EnvironmentRegistry.layer.pipe(Layer.provideMerge(dependencies))), diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index f6c305ab6bb90..d9893693ddad9 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -14,14 +14,17 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Scheduler from "effect/Scheduler"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import * as Tracer from "effect/Tracer"; +import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; import * as TokenStore from "../authorization/tokenStore.ts"; import { @@ -167,6 +170,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( target: ConnectionTarget, ) => Effect.Effect; readonly initialDisabled?: ReadonlyArray; + readonly initialLastConnected?: ReadonlyArray<{ + readonly environmentId: EnvironmentId; + readonly atEpochMs: number; + }>; + readonly onDpopPrefetch?: (environmentId: EnvironmentId) => Effect.Effect; readonly directEndpoints?: ServerConfig["directEndpoints"]; }, ) { @@ -205,9 +213,21 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( const storedDisabled = yield* Ref.make>( new Set(options?.initialDisabled ?? []), ); + const storedLastConnected = yield* Ref.make( + new Map( + (options?.initialLastConnected ?? []).map( + (row) => [row.environmentId, row.atEpochMs] as const, + ), + ), + ); const targetStore = Persistence.ConnectionTargetStore.of({ list: Ref.get(storedTargets), listDisabled: Ref.get(storedDisabled).pipe(Effect.map((ids) => [...ids])), + listLastConnected: Ref.get(storedLastConnected).pipe( + Effect.map((rows) => + [...rows.entries()].map(([environmentId, atEpochMs]) => ({ environmentId, atEpochMs })), + ), + ), }); const registrationStore = Persistence.ConnectionRegistrationStore.of({ register: (registration, routes) => @@ -264,6 +284,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( next.delete(environmentId); return next; }); + yield* Ref.update(storedLastConnected, (current) => { + const next = new Map(current); + next.delete(environmentId); + return next; + }); }), setEnabled: (environmentId, enabled) => Ref.update(storedDisabled, (current) => { @@ -275,6 +300,12 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( } return next; }), + touchConnected: (environmentId, atEpochMs) => + Ref.update(storedLastConnected, (current) => { + const next = new Map(current); + next.set(environmentId, atEpochMs); + return next; + }), }); const cacheStore = Persistence.EnvironmentCacheStore.of({ loadShell: (environmentId) => @@ -315,6 +346,34 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( const connectivity = Connectivity.Connectivity.of({ status: SubscriptionRef.get(networkStatus), changes: SubscriptionRef.changes(networkStatus), + privateNetworkLikely: Effect.succeed(true), + }); + const dpopPrefetchCalls = yield* Ref.make(0); + const wakeups = yield* SubscriptionRef.make<{ + readonly sequence: number; + readonly reason: ConnectionWakeups.ConnectionWakeup; + }>({ + sequence: 0, + reason: "application-active", + }); + const remoteAuthorization = RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization.of({ + authorizeBearer: () => Effect.die("unused"), + authorizeDpop: () => Effect.die("unused"), + authorizeDpopHttp: (input) => + Ref.update(dpopPrefetchCalls, (count) => count + 1).pipe( + Effect.andThen(options?.onDpopPrefetch?.(input.expectedEnvironmentId) ?? Effect.void), + Effect.as({ + environmentId: input.expectedEnvironmentId, + label: "Prefetched", + httpBaseUrl: "https://relay.example.test", + httpAuthorization: { + _tag: "Dpop" as const, + accessToken: "prefetch-token", + expiresAtEpochMs: Number.MAX_SAFE_INTEGER, + }, + }), + ), + renewDpopBeforeExpiry: () => Effect.die("unused"), }); const profileStore = ConnectionProfileStore.ConnectionProfileStore.of({ get: (connectionId) => @@ -436,12 +495,21 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( return { prepared, session }; }); const driver = ConnectionDriver.ConnectionDriver.of({ - connect: (entry, reportProgress) => + connect: (entry, reportProgress, options) => reportProgress({ stage: "preparing" }).pipe( Effect.andThen( - ConnectionDriver.connectOverRoutes(entry, checkRoute, (route) => - connectRoute(entry, route), - ), + Effect.gen(function* () { + return yield* ConnectionDriver.connectOverRoutes( + entry, + checkRoute, + (route) => connectRoute(entry, route), + { + privateNetworkLikely: + options?.privateNetworkLikely ?? (yield* connectivity.privateNetworkLikely), + ...(options?.hedge === undefined ? {} : { hedge: options.hedge }), + }, + ); + }), ), ), preflight: (_entry, route) => @@ -466,7 +534,16 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( Layer.succeed(Connectivity.Connectivity, connectivity), Layer.succeed( ConnectionWakeups.ConnectionWakeups, - ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.never }), + ConnectionWakeups.ConnectionWakeups.of({ + changes: SubscriptionRef.changes(wakeups).pipe( + Stream.drop(1), + Stream.map((event) => event.reason), + ), + }), + ), + Layer.succeed( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + remoteAuthorization, ), Layer.succeed(ConnectionDriver.ConnectionDriver, driver), layerCache, @@ -488,9 +565,16 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( storedCredentials, storedRemoteTokens, storedDisabled, + storedLastConnected, disconnectedSshTargets, networkStatus, connectedRoutes, + dpopPrefetchCalls, + wake: (reason: ConnectionWakeups.ConnectionWakeup) => + SubscriptionRef.update(wakeups, (event) => ({ + sequence: event.sequence + 1, + reason, + })), }; }); @@ -622,6 +706,91 @@ describe("EnvironmentRegistry", () => { }), ); + it.effect("defers cold start for environments last reached hours ago", () => + Effect.gen(function* () { + const started = yield* Ref.make>([]); + const freshStarted = yield* Deferred.make(); + const staleStarted = yield* Deferred.make(); + const releaseLoads = yield* Deferred.make(); + const harness = yield* makeHarness([TARGET, SECOND_TARGET], [], [], { + initialLastConnected: [{ environmentId: SECOND_TARGET.environmentId, atEpochMs: 0 }], + beforeSessionConnect: (environmentId) => + Ref.update(started, (current) => [...current, environmentId]).pipe( + Effect.andThen( + environmentId === TARGET.environmentId + ? Deferred.succeed(freshStarted, undefined) + : Deferred.succeed(staleStarted, undefined), + ), + Effect.andThen(Deferred.await(releaseLoads)), + ), + }); + + yield* TestClock.adjust("6 hours"); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* Effect.forkChild(registry.start); + + yield* Deferred.await(freshStarted); + expect(yield* Ref.get(started)).toEqual([TARGET.environmentId]); + + yield* TestClock.adjust("2 seconds"); + yield* Deferred.await(staleStarted); + expect(yield* Ref.get(started)).toEqual([ + TARGET.environmentId, + SECOND_TARGET.environmentId, + ]); + + yield* Deferred.succeed(releaseLoads, undefined); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + + it.effect("records when an environment last connected successfully", () => + Effect.gen(function* () { + const harness = yield* makeHarness([TARGET]); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.start; + yield* awaitConnectionState( + registry, + TARGET.environmentId, + (state) => state.phase === "connected", + ); + + expect(yield* Ref.get(harness.storedLastConnected)).toEqual( + new Map([[TARGET.environmentId, 0]]), + ); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + + it.effect( + "prefetches T3 Connect credentials on start and wake for recent relay environments", + () => + Effect.gen(function* () { + const prefetches = yield* Queue.unbounded(); + const harness = yield* makeHarness([RELAY_TARGET], [], [], { + onDpopPrefetch: (environmentId) => Queue.offer(prefetches, environmentId), + }); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.start; + expect(yield* Queue.take(prefetches)).toBe(RELAY_TARGET.environmentId); + yield* awaitConnectionState( + registry, + RELAY_TARGET.environmentId, + (state) => state.phase === "connected", + ); + + yield* harness.wake("application-active"); + expect(yield* Queue.take(prefetches)).toBe(RELAY_TARGET.environmentId); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("does not expose a persisted environment as disconnected while it starts", () => Effect.gen(function* () { const connectionStarted = yield* Deferred.make(); @@ -2196,6 +2365,39 @@ describe("EnvironmentRegistry routes", () => { }), ); + it.effect("remembers the successful route first without dropping the live session", () => + Effect.gen(function* () { + const harness = yield* makeHarness( + [LAN_TARGET, RELAY_TARGET], + [LAN_PROFILE], + [[LAN_TARGET.connectionId, BEARER_CREDENTIAL]], + { + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" ? "silent" : "unchecked", + }, + ); + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.start; + yield* awaitConnectionState( + registry, + LAN_TARGET.environmentId, + (state) => state.phase === "connected", + ); + + expect(routesOf(yield* Ref.get(harness.storedTargets), LAN_TARGET.environmentId)).toEqual([ + RELAY_TARGET, + LAN_TARGET, + ]); + expect(yield* Ref.get(harness.connectedRoutes)).toEqual(["relay"]); + expect(yield* Ref.get(harness.releasedSessions)).toBe(0); + const entry = (yield* SubscriptionRef.get(registry.entries)).get(LAN_TARGET.environmentId); + expect(entry?.target).toEqual(RELAY_TARGET); + expect(entry?.alternateRoutes?.map((route) => route.target)).toEqual([LAN_TARGET]); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("removing the last route forgets the environment", () => Effect.gen(function* () { const harness = yield* makeHarness( diff --git a/packages/client-runtime/src/connection/registry.ts b/packages/client-runtime/src/connection/registry.ts index 12f8aa5ec505f..fdff36bbb34bc 100644 --- a/packages/client-runtime/src/connection/registry.ts +++ b/packages/client-runtime/src/connection/registry.ts @@ -1,5 +1,7 @@ import { EnvironmentId } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; import * as Exit from "effect/Exit"; @@ -12,6 +14,7 @@ import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; import { type ConnectionCatalogEntry, @@ -33,7 +36,7 @@ import type { PersistedConnectionTarget, SupervisorConnectionState, } from "./model.ts"; -import { ConnectionBlockedError } from "./model.ts"; +import { ConnectionBlockedError, DPOP_WAKE_REFRESH_SKEW_MS } from "./model.ts"; import * as Persistence from "../platform/persistence.ts"; import * as EnvironmentSupervisor from "./supervisor.ts"; import * as ConnectionDriver from "./driver.ts"; @@ -47,10 +50,12 @@ import { connectionRouteId, connectionRoutes, credentialConnectionId, + entryUsesDpop, entryWithRoutes, findRouteToSameAddress, hasSavedBearerRoute, mergeLearnedRoutes, + preferRoute, routesAfterRemoving, upsertRoute, type ReportedEndpoint, @@ -58,6 +63,11 @@ import { const isSshConnectionProfile = Schema.is(SshConnectionProfile); +/** Environments last reached longer ago than this wait until after recent ones on cold start. */ +const STALE_RECONNECT_AFTER_MS = 6 * 60 * 60 * 1000; +/** Lets recently used environments open sockets before stale ones compete. */ +const STALE_START_DELAY = Duration.seconds(2); + function unsupportedState( entry: ConnectionCatalogEntry, ): Pick { @@ -219,6 +229,7 @@ export const make = Effect.gen(function* () { const connectivity = yield* Connectivity.Connectivity; const driver = yield* ConnectionDriver.ConnectionDriver; const wakeups = yield* ConnectionWakeups.ConnectionWakeups; + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; const ssh = yield* ClientCapabilities.SshEnvironmentGateway; const persistedTargets = yield* storage.list; const disabledEnvironmentIds = new Set(yield* storage.listDisabled); @@ -368,6 +379,7 @@ export const make = Effect.gen(function* () { // before an enabled, compatible environment starts connecting. initiallyDesired: entry.enabled && entry.unsupportedReason === undefined, learnRoutes: (input) => learnRoutes({ environmentId, ...input }), + rememberRoute: (input) => rememberSuccessfulRoute({ environmentId, ...input }), }).pipe( Effect.provideService(Connectivity.Connectivity, connectivity), Effect.provideService(ConnectionDriver.ConnectionDriver, driver), @@ -468,21 +480,93 @@ export const make = Effect.gen(function* () { ), ); + const startEnvironment = (environmentId: EnvironmentId) => + acquireSupervisor(environmentId).pipe( + Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }), + ); + + const loadLastConnectedById = Effect.fnUntraced(function* () { + return new Map( + (yield* storage.listLastConnected.pipe( + Effect.orElseSucceed( + (): ReadonlyArray<{ environmentId: EnvironmentId; atEpochMs: number }> => [], + ), + )).map((row) => [row.environmentId, row.atEpochMs] as const), + ); + }); + + /** + * Warms T3 Connect access tokens before (or while) supervisors open sockets, + * so prepare often hits the token cache instead of a full bootstrap exchange. + */ + const prefetchDpopCredentials = Effect.fn("EnvironmentRegistry.prefetchDpopCredentials")( + function* () { + const now = yield* Clock.currentTimeMillis; + const lastConnectedById = yield* loadLastConnectedById(); + const current = yield* SubscriptionRef.get(entries); + yield* Effect.forEach( + current.values(), + (entry) => { + if (!entry.enabled || entry.unsupportedReason !== undefined || !entryUsesDpop(entry)) { + return Effect.void; + } + const lastConnectedAt = lastConnectedById.get(entry.target.environmentId); + if (lastConnectedAt !== undefined && now - lastConnectedAt >= STALE_RECONNECT_AFTER_MS) { + return Effect.void; + } + return remote + .authorizeDpopHttp({ + expectedEnvironmentId: entry.target.environmentId, + refreshSkewMs: DPOP_WAKE_REFRESH_SKEW_MS, + }) + .pipe( + Effect.asVoid, + Effect.catch((error) => + Effect.logDebug("Could not prefetch T3 Connect credentials.", { + environmentId: entry.target.environmentId, + error, + }), + ), + ); + }, + { concurrency: 2, discard: true }, + ); + }, + ); + const start = Effect.gen(function* () { if (yield* Ref.getAndSet(started, true)) { return; } + const now = yield* Clock.currentTimeMillis; + const lastConnectedById = yield* loadLastConnectedById(); + const fresh: EnvironmentId[] = []; + const stale: EnvironmentId[] = []; + for (const environmentId of persistedRoutesByEnvironment.keys()) { + const lastConnectedAt = lastConnectedById.get(environmentId); + if (lastConnectedAt !== undefined && now - lastConnectedAt >= STALE_RECONNECT_AFTER_MS) { + stale.push(environmentId); + } else { + // Never-connected environments stay immediate so first pairing is not delayed. + fresh.push(environmentId); + } + } + // Overlap token mint with supervisor connect for recently used T3 Connect envs. + yield* prefetchDpopCredentials().pipe(Effect.forkIn(registryScope)); + yield* Effect.forEach(fresh, startEnvironment, { + concurrency: "unbounded", + discard: true, + }); + if (stale.length === 0) return; yield* Effect.forEach( - persistedRoutesByEnvironment.keys(), + stale, (environmentId) => - acquireSupervisor(environmentId).pipe( - Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }), - ), + Effect.sleep(STALE_START_DELAY).pipe(Effect.andThen(startEnvironment(environmentId))), { concurrency: "unbounded", discard: true, }, - ); + ).pipe(Effect.forkIn(registryScope)); }).pipe(Effect.withSpan("EnvironmentRegistry.start")); const installEntryLocked = Effect.fn("EnvironmentRegistry.installEntryLocked")(function* ( @@ -915,6 +999,68 @@ export const make = Effect.gen(function* () { const allowInsecureRoutes = typeof globalThis.location === "undefined" || globalThis.location.protocol !== "https:"; + /** + * Persist route order without restarting the live supervisor. `installEntryLocked` + * would tear the session down; learned and sticky updates only need the catalog. + */ + const publishRoutesLocked = Effect.fn("EnvironmentRegistry.publishRoutesLocked")(function* ( + environmentId: EnvironmentId, + next: ConnectionCatalogEntry, + ) { + yield* registrations.setRoutes(environmentId, persistedRoutes(next)); + const lease = (yield* SubscriptionRef.get(serviceScopes)).get(environmentId); + if (lease !== undefined) { + yield* SubscriptionRef.update(serviceScopes, (current) => + new Map(current).set(environmentId, { ...lease, entry: next }), + ); + } + yield* SubscriptionRef.update(entries, (current) => new Map(current).set(environmentId, next)); + }); + + /** + * The route that just connected moves first so the next cold start tries it + * before a dead LAN or other slower path from the previous network. + */ + const rememberSuccessfulRoute = Effect.fn("EnvironmentRegistry.rememberSuccessfulRoute")( + function* (input: { + readonly environmentId: EnvironmentId; + readonly activeRoute: ConnectionRoute; + }) { + return yield* withLeaseLock( + input.environmentId, + Effect.gen(function* () { + if ((yield* Ref.get(platformEnvironmentIds)).has(input.environmentId)) { + return Option.none(); + } + const entry = (yield* SubscriptionRef.get(entries)).get(input.environmentId); + if (entry === undefined) return Option.none(); + const now = yield* Clock.currentTimeMillis; + yield* registrations.touchConnected(input.environmentId, now).pipe( + Effect.catch((error) => + Effect.logWarning("Could not record the last connection time.", { + environmentId: input.environmentId, + error, + }), + ), + ); + const current = connectionRoutes(entry); + const routes = preferRoute(current, connectionRouteId(input.activeRoute.target)); + if (routes === current) return Option.none(); + const next = entryWithRoutes(entry, routes); + yield* publishRoutesLocked(input.environmentId, next); + return Option.some(next); + }), + ).pipe( + Effect.catch((error) => + Effect.logWarning("Could not remember the successful connection route.", { + environmentId: input.environmentId, + error, + }).pipe(Effect.as(Option.none())), + ), + ); + }, + ); + /** * Learned routes reuse the active route's credential, so they do not revoke pull request routing * trust the way a newly paired address does. @@ -956,16 +1102,7 @@ export const make = Effect.gen(function* () { } yield* profiles.put(profile); } - yield* registrations.setRoutes(input.environmentId, persistedRoutes(next)); - const lease = (yield* SubscriptionRef.get(serviceScopes)).get(input.environmentId); - if (lease !== undefined) { - yield* SubscriptionRef.update(serviceScopes, (current) => - new Map(current).set(input.environmentId, { ...lease, entry: next }), - ); - } - yield* SubscriptionRef.update(entries, (current) => - new Map(current).set(input.environmentId, next), - ); + yield* publishRoutesLocked(input.environmentId, next); return Option.some(next); }), ).pipe( @@ -1139,6 +1276,17 @@ export const make = Effect.gen(function* () { Stream.runForEach((status) => SubscriptionRef.set(networkStatus, status)), Effect.forkScoped, ); + yield* wakeups.changes.pipe( + Stream.filter( + (reason) => + reason === "application-active" || + reason === "application-active-probe" || + reason === "application-active-long" || + reason === "network-changed", + ), + Stream.runForEach(() => prefetchDpopCredentials()), + Effect.forkScoped, + ); const setCompatibility = Effect.fn("EnvironmentRegistry.setCompatibility")(function* ( environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/connection/routes.test.ts b/packages/client-runtime/src/connection/routes.test.ts index da9c631638cc4..572649e7bb5cb 100644 --- a/packages/client-runtime/src/connection/routes.test.ts +++ b/packages/client-runtime/src/connection/routes.test.ts @@ -18,8 +18,12 @@ import { entryWithRoutes, mergeLearnedRoutes, pendingRouteOrder, + preferRoute, removedWithRelay, + routeUsesDpop, routesAfterRemoving, + routesForNetwork, + routesIncludeLan, sendsTokenInClear, upsertRoute, usableRoutes, @@ -84,6 +88,37 @@ describe("connection routes", () => { expect(insertRoute([LAN, TAILNET], RELAY)).toEqual([LAN, TAILNET, RELAY]); }); + it("moves a successful route to the front for the next cold start", () => { + const routes = [LAN, TAILNET, RELAY]; + expect(preferRoute(routes, "relay")).toEqual([RELAY, LAN, TAILNET]); + expect(preferRoute(routes, "lan")).toBe(routes); + expect(preferRoute(routes, "missing")).toBe(routes); + }); + + it("skips LAN routes when the device is not on a private network", () => { + const entry: ConnectionCatalogEntry = { + target: LAN.target, + profile: LAN.profile, + alternateRoutes: [TAILNET, RELAY], + enabled: true, + }; + expect(routesForNetwork(entry, true)).toEqual([LAN, TAILNET, RELAY]); + expect(routesForNetwork(entry, false)).toEqual([TAILNET, RELAY]); + expect(routesIncludeLan([LAN, TAILNET])).toBe(true); + expect(routesIncludeLan([TAILNET, RELAY])).toBe(false); + expect(routeUsesDpop(RELAY)).toBe(true); + expect(routeUsesDpop(LAN)).toBe(false); + }); + + it("keeps LAN when it is the only saved route", () => { + const entry: ConnectionCatalogEntry = { + target: LAN.target, + profile: LAN.profile, + enabled: true, + }; + expect(routesForNetwork(entry, false)).toEqual([LAN]); + }); + it("keeps a user's order when a saved route is replaced", () => { const repaired = direct("lan", "http://192.168.1.11:3773/"); expect(upsertRoute([RELAY, LAN], repaired)).toEqual([RELAY, repaired]); diff --git a/packages/client-runtime/src/connection/routes.ts b/packages/client-runtime/src/connection/routes.ts index 9d83550a8c4a2..5b6e09930b3c6 100644 --- a/packages/client-runtime/src/connection/routes.ts +++ b/packages/client-runtime/src/connection/routes.ts @@ -140,6 +140,17 @@ export function upsertRoute( insertRoute(routes, route, { aheadOfSameKind: true }); } +/** Moves a saved route to the front so the next cold start tries it first. */ +export function preferRoute( + routes: ReadonlyArray, + routeId: string, +): ReadonlyArray { + const index = routes.findIndex((route) => connectionRouteId(route.target) === routeId); + if (index <= 0) return routes; + const preferred = routes[index]!; + return [preferred, ...routes.slice(0, index), ...routes.slice(index + 1)]; +} + export function findRouteToSameAddress( routes: ReadonlyArray, route: ConnectionRoute, @@ -356,6 +367,35 @@ export function usableRoutes(entry: ConnectionCatalogEntry): ReadonlyArray !sendsTokenInClear(routes, route)); } +/** + * Drops LAN routes when the device is not on a private network, unless that + * would leave the environment with no route at all. + */ +export function routesForNetwork( + entry: ConnectionCatalogEntry, + privateNetworkLikely: boolean, +): ReadonlyArray { + const routes = usableRoutes(entry); + if (privateNetworkLikely) return routes; + const withoutLan = routes.filter((route) => connectionRouteKind(route) !== "lan"); + return withoutLan.length > 0 ? withoutLan : routes; +} + +/** T3 Connect relay and learned/saved t3-connect bearer routes need a DPoP access token. */ +export function routeUsesDpop(route: ConnectionRoute): boolean { + if (route.target._tag === "RelayConnectionTarget") return true; + const profile = Option.getOrNull(route.profile); + return profile?._tag === "BearerConnectionProfile" && profile.authorization === "t3-connect"; +} + +export function entryUsesDpop(entry: ConnectionCatalogEntry): boolean { + return connectionRoutes(entry).some(routeUsesDpop); +} + +export function routesIncludeLan(routes: ReadonlyArray): boolean { + return routes.some((route) => connectionRouteKind(route) === "lan"); +} + export interface ReportedEndpoint { readonly httpBaseUrl: string; readonly kind?: string; diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index a90f3f0e3ab2a..a231b11407299 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -165,6 +165,7 @@ function recordTransitions(supervisor: EnvironmentSupervisor.EnvironmentSupervis const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: { readonly networkStatus?: NetworkStatus; + readonly privateNetworkLikely?: boolean; readonly prepare?: ( attempt: number, target: ConnectionTarget, @@ -195,9 +196,11 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: ReadonlyArray> >([]); + const privateNetworkLikely = yield* Ref.make(options?.privateNetworkLikely ?? true); const connectivity = Connectivity.Connectivity.of({ status: SubscriptionRef.get(networkStatus), changes: SubscriptionRef.changes(networkStatus), + privateNetworkLikely: Ref.get(privateNetworkLikely), }); const prepare = Effect.fn("TestConnectionDriver.prepare")(function* (target: ConnectionTarget) { @@ -251,10 +254,18 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: const connect = Effect.fn("TestConnectionDriver.connect")(function* ( entry: ConnectionCatalogEntry, reportProgress: (progress: ConnectionDriver.ConnectionDriverProgress) => Effect.Effect, + options?: ConnectionDriver.ConnectOverRoutesOptions, ) { yield* reportProgress({ stage: "preparing" }); - return yield* ConnectionDriver.connectOverRoutes(entry, checkRoute, (route) => - connectRoute(route.target, reportProgress), + const onPrivateNetwork = yield* Ref.get(privateNetworkLikely); + return yield* ConnectionDriver.connectOverRoutes( + entry, + checkRoute, + (route) => connectRoute(route.target, reportProgress), + { + privateNetworkLikely: onPrivateNetwork, + ...(options?.hedge === undefined ? {} : { hedge: options.hedge }), + }, ); }); @@ -1752,6 +1763,118 @@ describe("EnvironmentSupervisor routes", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("hedges to relay when a ready LAN route hangs on connect", () => + Effect.gen(function* () { + const checkingLan = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Deferred.succeed(checkingLan, undefined).pipe( + Effect.andThen(Effect.sleep("100 millis")), + Effect.as("answered" as const), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => + target._tag === "BearerConnectionTarget" + ? Effect.never + : Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(checkingLan); + yield* TestClock.adjust("100 millis"); + // Head-start classification finishes, LAN connect starts, then the hedge opens relay. + yield* TestClock.adjust("250 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "RelayConnectionTarget", + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(0); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("releases the losing hedged session when both routes connect", () => + Effect.gen(function* () { + const checkingLan = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Deferred.succeed(checkingLan, undefined).pipe( + Effect.andThen(Effect.sleep("100 millis")), + Effect.as("answered" as const), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + // LAN acquires a session then stalls in ready so the relay hedge can + // win; interrupting that open session must release it. + ready: (attempt) => (attempt === 1 ? Effect.sleep("1 second") : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(checkingLan); + yield* TestClock.adjust("100 millis"); + yield* TestClock.adjust("250 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "RelayConnectionTarget", + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("skips LAN when the device is not on a private network", () => + Effect.gen(function* () { + const preparedTargets = yield* Ref.make>([]); + const harness = yield* makeHarness({ + privateNetworkLikely: false, + prepare: (_attempt, target) => + Ref.update(preparedTargets, (current) => [...current, target]).pipe( + Effect.as(preparedFor(target)), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(yield* Ref.get(preparedTargets)).toEqual([RELAY_TARGET]); + }), + ); + + it.effect("does not wait for the LAN head-start when only remote routes remain", () => + Effect.gen(function* () { + const checking = yield* Ref.make(0); + const preparedTargets = yield* Ref.make>([]); + const harness = yield* makeHarness({ + privateNetworkLikely: false, + checkRoute: () => + Ref.update(checking, (count) => count + 1).pipe( + Effect.andThen(Effect.sleep("10 seconds")), + Effect.as("answered" as const), + ), + prepare: (_attempt, target) => + Ref.update(preparedTargets, (current) => [...current, target]).pipe( + Effect.as(preparedFor(target)), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + // Without a LAN candidate the driver skips reachability classification entirely. + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(yield* Ref.get(checking)).toBe(0); + expect(yield* Ref.get(preparedTargets)).toEqual([RELAY_TARGET]); + }), + ); + it.effect("tries a slow tailnet route that answered before a dead LAN route", () => Effect.gen(function* () { const tailnetTarget = new BearerConnectionTarget({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index b2559f123b2eb..1430cad92df2a 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -165,6 +165,10 @@ export interface EnvironmentSupervisorOptions { readonly activeRoute: ConnectionRoute; readonly reported: ReadonlyArray; }) => Effect.Effect>; + /** Persist the route that connected so the next cold start tries it first. */ + readonly rememberRoute?: (input: { + readonly activeRoute: ConnectionRoute; + }) => Effect.Effect>; } function retryDelayMs(failureCount: number, random: number): number { @@ -520,7 +524,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( preferredRoute: Option.Option, report: ProgressReporter, ) { - return yield* driver.connect(yield* attemptEntry(preferredRoute), report); + // Route upgrades must try the preferred path serially. Hedging would open the + // live route as a backup and look like a successful upgrade on the wrong path. + return yield* driver.connect(yield* attemptEntry(preferredRoute), report, { + hedge: Option.isNone(preferredRoute), + }); }); // Low-cardinality dimensions for time-to-connected dashboards; the error @@ -1275,6 +1283,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }; const connectedAt = yield* Clock.currentTimeMillis; yield* setLease(connectedLease); + if (connectedLease.activeRoute !== undefined && options?.rememberRoute !== undefined) { + // Persist sticky order for the next cold start only. Applying it to the + // live entry would make the active route look preferred and stop + // better-route upgrades (for example LAN while connected over relay). + yield* options.rememberRoute({ activeRoute: connectedLease.activeRoute }); + } if (options?.learnRoutes !== undefined) { yield* learnRoutesFrom(options.learnRoutes, connectedLease).pipe(Effect.forkScoped); } diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index e043279d3e576..9015b1629e1fc 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -21,10 +21,12 @@ export class ConnectionPersistenceError extends Schema.TaggedError, ConnectionPersistenceError>; readonly listDisabled: Effect.Effect, ConnectionPersistenceError>; + readonly listLastConnected: Effect.Effect< + ReadonlyArray<{ readonly environmentId: EnvironmentId; readonly atEpochMs: number }>, + ConnectionPersistenceError + >; } >()("@t3tools/client-runtime/platform/persistence/ConnectionTargetStore") {} @@ -74,6 +80,10 @@ export class ConnectionRegistrationStore extends Context.Service< environmentId: EnvironmentId, enabled: boolean, ) => Effect.Effect; + readonly touchConnected: ( + environmentId: EnvironmentId, + atEpochMs: number, + ) => Effect.Effect; } >()("@t3tools/client-runtime/platform/persistence/ConnectionRegistrationStore") {} diff --git a/packages/client-runtime/src/platform/storageDocument.test.ts b/packages/client-runtime/src/platform/storageDocument.test.ts index 60f561e3dae53..a10ea593db90a 100644 --- a/packages/client-runtime/src/platform/storageDocument.test.ts +++ b/packages/client-runtime/src/platform/storageDocument.test.ts @@ -33,6 +33,7 @@ import { registerConnectionInCatalog, removeConnectionFromCatalog, setConnectionEnabledInCatalog, + touchEnvironmentConnectedInCatalog, } from "./storageDocument.ts"; const decodeConnectionCatalogDocument = Schema.decodeUnknownEffect(ConnectionCatalogDocument); @@ -240,6 +241,30 @@ describe("ConnectionCatalogDocument", () => { ); }); + it("records and clears last-connected timestamps with the environment", () => { + const registered = registerConnectionInCatalog( + EMPTY_CONNECTION_CATALOG_DOCUMENT, + new BearerConnectionRegistration({ + target: BEARER_TARGET, + profile: BEARER_PROFILE, + credential: BEARER_CREDENTIAL, + }), + ); + const touched = touchEnvironmentConnectedInCatalog( + registered, + ENVIRONMENT_ID, + 1_700_000_000_000, + ); + + expect(touched.lastConnected).toEqual([ + { environmentId: ENVIRONMENT_ID, atEpochMs: 1_700_000_000_000 }, + ]); + expect(removeConnectionFromCatalog(touched, ENVIRONMENT_ID).lastConnected).toEqual([]); + expect( + touchEnvironmentConnectedInCatalog(EMPTY_CONNECTION_CATALOG_DOCUMENT, ENVIRONMENT_ID, 1), + ).toBe(EMPTY_CONNECTION_CATALOG_DOCUMENT); + }); + it("stores a DPoP token for a registered relay environment", () => { const registered = registerConnectionInCatalog( EMPTY_CONNECTION_CATALOG_DOCUMENT, diff --git a/packages/client-runtime/src/platform/storageDocument.ts b/packages/client-runtime/src/platform/storageDocument.ts index 1af8e0e3d67aa..4940013db76d6 100644 --- a/packages/client-runtime/src/platform/storageDocument.ts +++ b/packages/client-runtime/src/platform/storageDocument.ts @@ -17,6 +17,12 @@ export const StoredConnectionCredential = Schema.Struct({ }); export type StoredConnectionCredential = typeof StoredConnectionCredential.Type; +export const StoredEnvironmentLastConnected = Schema.Struct({ + environmentId: EnvironmentId, + atEpochMs: Schema.Number, +}); +export type StoredEnvironmentLastConnected = typeof StoredEnvironmentLastConnected.Type; + export const ConnectionCatalogDocument = Schema.Struct({ schemaVersion: Schema.Literal(1), targets: Schema.Array(PersistedConnectionTarget), @@ -29,6 +35,10 @@ export const ConnectionCatalogDocument = Schema.Struct({ disabledEnvironmentIds: Schema.Array(EnvironmentId).pipe( Schema.withDecodingDefaultKey(Effect.succeed([])), ), + // Cold start uses this to connect recently used environments before stale ones. + lastConnected: Schema.Array(StoredEnvironmentLastConnected).pipe( + Schema.withDecodingDefaultKey(Effect.succeed([])), + ), }); export type ConnectionCatalogDocument = typeof ConnectionCatalogDocument.Type; @@ -39,6 +49,7 @@ export const EMPTY_CONNECTION_CATALOG_DOCUMENT: ConnectionCatalogDocument = Obje credentials: [], remoteDpopTokens: [], disabledEnvironmentIds: [], + lastConnected: [], }); export function replaceCatalogValue( @@ -168,6 +179,11 @@ export function removeConnectionFromCatalog( (value) => value, environmentId, ), + lastConnected: removeCatalogValue( + next.lastConnected, + (value) => value.environmentId, + environmentId, + ), ...(next.githubRoutingPermissions === undefined ? {} : { @@ -178,6 +194,23 @@ export function removeConnectionFromCatalog( }; } +export function touchEnvironmentConnectedInCatalog( + document: ConnectionCatalogDocument, + environmentId: EnvironmentId, + atEpochMs: number, +): ConnectionCatalogDocument { + if (!document.targets.some((target) => target.environmentId === environmentId)) { + return document; + } + return { + ...document, + lastConnected: replaceCatalogValue(document.lastConnected, (value) => value.environmentId, { + environmentId, + atEpochMs, + }), + }; +} + export function setConnectionEnabledInCatalog( document: ConnectionCatalogDocument, environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/relay/discovery.test.ts b/packages/client-runtime/src/relay/discovery.test.ts index 9332652df39b5..27673b323bcb4 100644 --- a/packages/client-runtime/src/relay/discovery.test.ts +++ b/packages/client-runtime/src/relay/discovery.test.ts @@ -127,6 +127,7 @@ const makeHarness = Effect.fn("RelayDiscoveryTest.makeHarness")(function* () { const connectivity = Connectivity.Connectivity.of({ status: SubscriptionRef.get(networkStatus), changes: SubscriptionRef.changes(networkStatus), + privateNetworkLikely: Effect.succeed(true), }); const layer = RelayEnvironmentDiscovery.layer.pipe( Layer.provide( @@ -381,6 +382,7 @@ describe("RelayEnvironmentDiscovery", () => { Layer.succeed(Connectivity.Connectivity, { status: SubscriptionRef.get(networkStatus), changes: SubscriptionRef.changes(networkStatus), + privateNetworkLikely: Effect.succeed(true), }), Layer.succeed( ConnectionWakeups.ConnectionWakeups, diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 8f4bee5dc16c6..0fdd1103eb753 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -277,13 +277,16 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents }); describe("RpcSessionFactory", () => { - it.effect("connects while a large config snapshot downloads at 4000 bytes per second", () => + it.effect("becomes ready when the socket opens, before a large config snapshot finishes", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); + const configFiber = yield* Effect.forkChild(session.initialConfig); const socket = yield* awaitSocket(sockets); socket.open(); + yield* Fiber.join(readyFiber); + const request = yield* awaitRequest(socket); // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair. const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) }; @@ -301,9 +304,7 @@ describe("RpcSessionFactory", () => { yield* TestClock.adjust((bytes.length / 4000) * 1000); socket.serverMessage(new TextDecoder().decode(bytes)); } - yield* Fiber.join(readyFiber); - - expect(yield* session.initialConfig).toEqual(config); + expect(yield* Fiber.join(configFiber)).toEqual(config); }), ); @@ -966,7 +967,9 @@ describe("RpcSessionFactory", () => { Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); - const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready)); + const readyFiber = yield* Effect.forkChild(session.ready); + const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed)); + const initialConfigFiber = yield* Effect.forkChild(Effect.flip(session.initialConfig)); const configFiber = yield* session .subscribeServerConfig({}) .pipe(Stream.runHead, Effect.flip, Effect.forkChild); @@ -975,6 +978,7 @@ describe("RpcSessionFactory", () => { .pipe(Stream.runHead, Effect.flip, Effect.forkChild); const socket = yield* awaitSocket(sockets); socket.open(); + yield* Fiber.join(readyFiber); yield* completeInitialConfig(socket, { ...ENCODED_SERVER_CONFIG, environment: { @@ -983,11 +987,15 @@ describe("RpcSessionFactory", () => { }, }); - const error = yield* Fiber.join(readyFiber); + const error = yield* Fiber.join(initialConfigFiber); expect(error).toMatchObject({ reason: "configuration", message: "Connected environment environment-2 does not match environment-1.", }); + expect(yield* Fiber.join(closedFiber)).toMatchObject({ + reason: "configuration", + message: "Connected environment environment-2 does not match environment-1.", + }); expect((yield* Fiber.join(configFiber))._tag).toBe("RpcClientError"); expect((yield* Fiber.join(customConfigFiber))._tag).toBe("RpcClientError"); }), diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 3ea4b73bf6bd4..a0e1a75750e57 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -55,6 +55,7 @@ export interface RpcSession { readonly subscribeServerConfig: ( input?: ServerConfigSubscriptionInput, ) => ServerConfigSubscription; + /** Resolves once the WebSocket is open. Prefer `initialConfig` when the snapshot is required. */ readonly ready: Effect.Effect; readonly probe: Effect.Effect; /** Frames received so far: proof the socket is alive that needs no reply. */ @@ -366,6 +367,12 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ), Effect.withSpan("environment.initialSync"), ); + // Ready no longer waits on this snapshot, so a mismatched or failed sync + // must still end the session; otherwise the supervisor stays "connected". + yield* initialConfig.pipe( + Effect.catch((error) => Deferred.fail(configSubscriptionClosed, error).pipe(Effect.asVoid)), + Effect.forkScoped, + ); const serverConfigEvents = Stream.unwrap( Effect.gen(function* () { const subscription = yield* PubSub.subscribe(serverConfigUpdates); @@ -441,8 +448,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( client: protocolClient, initialConfig, subscribeServerConfig, + // Transport is ready when the socket opens. Server config still arrives on + // `initialConfig` / `subscribeServerConfig`; waiting for the snapshot here + // kept the UI on "Connecting…" for the whole download on a cold host. ready: Deferred.await(connected).pipe( - Effect.andThen(initialConfig), Effect.asVoid, Effect.raceFirst(Deferred.await(disconnected)), ), From b8836e7e721f0ebd3a69d524b2cabe576d1adc83 Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Sat, 7 Nov 2026 11:46:27 -0800 Subject: [PATCH 3/4] fix(mobile): keep loading while a connected environment syncs Empty states were treating sibling sync errors as hard failures and flashing the disconnected UI while another environment was still connecting. Co-authored-by: Cursor --- apps/mobile/src/features/home/HomeScreen.tsx | 7 +- .../home/workspace-connection-status.test.ts | 153 ++++++++++++++++++ .../home/workspace-connection-status.ts | 28 +++- .../features/threads/NewTaskRouteScreen.tsx | 7 +- 4 files changed, 184 insertions(+), 11 deletions(-) diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index dac883341cc10..1235d0adcdaad 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -185,12 +185,13 @@ function deriveEmptyState(props: { } if ( - catalogState.hasConnectingEnvironment && !catalogState.hasLoadedShellSnapshot && - catalogState.connectionError === null + (catalogState.hasSyncingConnectedEnvironment || + catalogState.hasPendingShellSnapshot || + (catalogState.hasConnectingEnvironment && catalogState.connectionError === null)) ) { return { - title: "Connecting to environment", + title: catalogState.hasReadyEnvironment ? "Loading environment" : "Connecting to environment", detail: "Loading projects and threads from the saved environment.", loading: true, }; diff --git a/apps/mobile/src/features/home/workspace-connection-status.test.ts b/apps/mobile/src/features/home/workspace-connection-status.test.ts index abef14fd427fd..1020a7d2f4035 100644 --- a/apps/mobile/src/features/home/workspace-connection-status.test.ts +++ b/apps/mobile/src/features/home/workspace-connection-status.test.ts @@ -78,4 +78,157 @@ describe("workspace connection status", () => { showsProgress: true, }); }); + + it("keeps a quiet Updating label while reconnecting over a cached shell", () => { + const state = workspaceState({ + hasReadyEnvironment: false, + hasConnectingEnvironment: true, + hasLoadedShellSnapshot: true, + connectionState: "connecting", + connectingEnvironments: [ + { + environmentId: "environment-1" as never, + environmentLabel: "MacBook", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "connecting", + connectionError: null, + connectionErrorTraceId: null, + }, + ], + }); + + expect(workspaceConnectionStatusPresentation(state)).toEqual({ + label: "Updating...", + showsProgress: true, + }); + }); + + it("names the environment when reconnecting without a cached shell", () => { + const state = workspaceState({ + hasReadyEnvironment: false, + hasConnectingEnvironment: true, + hasLoadedShellSnapshot: false, + connectionState: "connecting", + connectingEnvironments: [ + { + environmentId: "environment-1" as never, + environmentLabel: "MacBook", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "connecting", + connectionError: null, + connectionErrorTraceId: null, + }, + ], + }); + + expect(workspaceConnectionStatusPresentation(state)).toEqual({ + label: "Reconnecting to MacBook", + showsProgress: true, + }); + }); + + it("names the environment when a cached reconnect has an error", () => { + const state = workspaceState({ + hasReadyEnvironment: false, + hasConnectingEnvironment: true, + hasLoadedShellSnapshot: true, + connectionState: "reconnecting", + connectionError: "Could not reach MacBook", + connectingEnvironments: [ + { + environmentId: "environment-1" as never, + environmentLabel: "MacBook", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "reconnecting", + connectionError: "Could not reach MacBook", + connectionErrorTraceId: null, + }, + ], + }); + + expect(workspaceConnectionStatusPresentation(state)).toEqual({ + label: "Reconnecting to MacBook", + showsProgress: false, + }); + }); + + it("ignores a failing second environment while a hopeful one is connecting over cache", () => { + const state = workspaceState({ + hasReadyEnvironment: false, + hasConnectingEnvironment: true, + hasLoadedShellSnapshot: true, + connectionState: "reconnecting", + connectionError: "Could not reach sleeping laptop", + connectingEnvironments: [ + { + environmentId: "environment-1" as never, + environmentLabel: "MacBook", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "connecting", + connectionError: null, + connectionErrorTraceId: null, + }, + { + environmentId: "environment-2" as never, + environmentLabel: "Sleeping laptop", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "reconnecting", + connectionError: "Could not reach sleeping laptop", + connectionErrorTraceId: null, + }, + ], + }); + + expect(workspaceConnectionStatusPresentation(state)).toEqual({ + label: "Updating...", + showsProgress: true, + }); + }); + + it("names only the hopeful environment when reconnecting without a cache", () => { + const state = workspaceState({ + hasReadyEnvironment: false, + hasConnectingEnvironment: true, + hasLoadedShellSnapshot: false, + connectionState: "reconnecting", + connectionError: "Could not reach sleeping laptop", + connectingEnvironments: [ + { + environmentId: "environment-1" as never, + environmentLabel: "MacBook", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "connecting", + connectionError: null, + connectionErrorTraceId: null, + }, + { + environmentId: "environment-2" as never, + environmentLabel: "Sleeping laptop", + displayUrl: "", + isRelayManaged: false, + isEnabled: true, + connectionState: "reconnecting", + connectionError: "Could not reach sleeping laptop", + connectionErrorTraceId: null, + }, + ], + }); + + expect(workspaceConnectionStatusPresentation(state)).toEqual({ + label: "Reconnecting to MacBook", + showsProgress: true, + }); + }); }); diff --git a/apps/mobile/src/features/home/workspace-connection-status.ts b/apps/mobile/src/features/home/workspace-connection-status.ts index 7bdb53b9b2832..6db4da49d942c 100644 --- a/apps/mobile/src/features/home/workspace-connection-status.ts +++ b/apps/mobile/src/features/home/workspace-connection-status.ts @@ -16,16 +16,32 @@ function shouldShowWorkspaceConnectionStatus(state: WorkspaceState): boolean { ); } +function hopefulConnectingEnvironments(state: WorkspaceState) { + // A sleeping second machine should not steal the header from one that may still connect. + const hopeful = state.connectingEnvironments.filter( + (environment) => environment.connectionError === null, + ); + return hopeful.length > 0 ? hopeful : state.connectingEnvironments; +} + function workspaceConnectionStatusLabel(state: WorkspaceState): string { if (state.networkStatus === "offline") return "You are offline"; if (state.hasReadyEnvironment && state.hasSyncingConnectedEnvironment) { return state.hasLoadedShellSnapshot ? "Syncing threads..." : "Loading threads..."; } - if (state.connectingEnvironments.length === 1) { - return `Reconnecting to ${state.connectingEnvironments[0]!.environmentLabel}`; + const connecting = hopefulConnectingEnvironments(state); + // Cached threads stay on screen; keep the header quiet while the socket catches up. + if ( + state.hasLoadedShellSnapshot && + connecting.some((environment) => environment.connectionError === null) + ) { + return "Updating..."; + } + if (connecting.length === 1) { + return `Reconnecting to ${connecting[0]!.environmentLabel}`; } - if (state.connectingEnvironments.length > 1) { - return `Reconnecting ${state.connectingEnvironments.length} environments`; + if (connecting.length > 1) { + return `Reconnecting ${connecting.length} environments`; } if (state.connectionError !== null) return state.connectionError; if (state.hasPendingShellSnapshot) { @@ -38,11 +54,13 @@ export function workspaceConnectionStatusPresentation( state: WorkspaceState, ): WorkspaceConnectionStatusPresentation | null { if (!shouldShowWorkspaceConnectionStatus(state)) return null; + const hopeful = hopefulConnectingEnvironments(state); + const hasHopefulConnect = hopeful.some((environment) => environment.connectionError === null); return { label: workspaceConnectionStatusLabel(state), showsProgress: state.networkStatus !== "offline" && - (state.hasReadyEnvironment || state.connectionError === null) && + (state.hasReadyEnvironment || state.connectionError === null || hasHopefulConnect) && (state.connectingEnvironments.length > 0 || state.hasPendingShellSnapshot), }; } diff --git a/apps/mobile/src/features/threads/NewTaskRouteScreen.tsx b/apps/mobile/src/features/threads/NewTaskRouteScreen.tsx index d904c3c67c543..b4baa98fb0fea 100644 --- a/apps/mobile/src/features/threads/NewTaskRouteScreen.tsx +++ b/apps/mobile/src/features/threads/NewTaskRouteScreen.tsx @@ -92,12 +92,13 @@ function deriveProjectEmptyState(catalogState: WorkspaceState): { } if ( - catalogState.hasConnectingEnvironment && !catalogState.hasLoadedShellSnapshot && - catalogState.connectionError === null + (catalogState.hasSyncingConnectedEnvironment || + catalogState.hasPendingShellSnapshot || + (catalogState.hasConnectingEnvironment && catalogState.connectionError === null)) ) { return { - title: "Connecting to environment", + title: catalogState.hasReadyEnvironment ? "Loading environment" : "Connecting to environment", detail: "Loading projects from the saved environment.", loading: true, }; From 85e77ebb85b9400c0eef2893189afb99db8a19ce Mon Sep 17 00:00:00 2001 From: wSedlacek Date: Wed, 11 Nov 2026 09:37:25 -0800 Subject: [PATCH 4/4] fix(client-runtime): remember sticky routes only after initial config Gate prefer-route persistence on a successful initialConfig so a socket that opens to the wrong environment cannot stick for the next cold start. Treat VPN/OTHER like UNKNOWN on mobile so Tailscale-on-Wi-Fi still keeps LAN routes. Co-authored-by: Cursor --- apps/mobile/src/connection/platform.ts | 3 ++ .../src/connection/registry.test.ts | 15 ++++--- .../src/connection/supervisor.test.ts | 42 +++++++++++++++++++ .../src/connection/supervisor.ts | 14 +++++-- third-party-licenses.config.json | 20 +++++++++ 5 files changed, 85 insertions(+), 9 deletions(-) diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index 5cac44ddf6a33..cfe435b3831d7 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -43,9 +43,12 @@ function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "on function privateNetworkLikely(state: Network.NetworkState): boolean { if (state.isConnected !== true) return false; + // VPN/OTHER often mean Wi‑Fi under a tunnel; keep LAN like UNKNOWN does. return ( state.type === Network.NetworkStateType.WIFI || state.type === Network.NetworkStateType.ETHERNET || + state.type === Network.NetworkStateType.VPN || + state.type === Network.NetworkStateType.OTHER || state.type === Network.NetworkStateType.UNKNOWN ); } diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 01130add393d7..b2a40e823726d 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -470,20 +470,21 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( yield* options?.beforeSessionConnect?.(target.environmentId) ?? Effect.void; const closed = yield* Deferred.make(); yield* Ref.update(sessions, (current) => [...current, { closed }]); + const sessionConfig = ( + options?.directEndpoints === undefined ? {} : { directEndpoints: options.directEndpoints } + ) as ServerConfig; const session = yield* Effect.acquireRelease( Effect.succeed({ client: {} as RpcSession.RpcSession["client"], - initialConfig: - options?.directEndpoints === undefined - ? Effect.die(new Error("Config is not used by registry tests.")) - : Effect.succeed({ directEndpoints: options.directEndpoints } as ServerConfig), + // Succeeds so sticky-route remember can wait on it like production. + initialConfig: Effect.succeed(sessionConfig), subscribeServerConfig: () => options?.directEndpoints === undefined ? Stream.never : Stream.make({ version: 1 as const, type: "snapshot" as const, - config: { directEndpoints: options.directEndpoints } as ServerConfig, + config: sessionConfig, }), ready: Effect.void, probe: Effect.void, @@ -759,6 +760,8 @@ describe("EnvironmentRegistry", () => { TARGET.environmentId, (state) => state.phase === "connected", ); + // rememberRoute forks after initialConfig; let that fiber finish. + yield* Effect.yieldNow; expect(yield* Ref.get(harness.storedLastConnected)).toEqual( new Map([[TARGET.environmentId, 0]]), @@ -2385,6 +2388,8 @@ describe("EnvironmentRegistry routes", () => { LAN_TARGET.environmentId, (state) => state.phase === "connected", ); + // rememberRoute forks after initialConfig; let that fiber finish. + yield* Effect.yieldNow; expect(routesOf(yield* Ref.get(harness.storedTargets), LAN_TARGET.environmentId)).toEqual([ RELAY_TARGET, diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index a231b11407299..c68bd2b0eda40 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -353,6 +353,48 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("remembers a sticky route only after initial config succeeds", () => + Effect.gen(function* () { + const releaseConfig = yield* Deferred.make(); + const remembered = yield* Deferred.make(); + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Deferred.await(releaseConfig), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + rememberRoute: ({ activeRoute }) => + Deferred.succeed(remembered, connectionRouteId(activeRoute.target)).pipe( + Effect.as(Option.none()), + ), + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.isNone(yield* Deferred.poll(remembered))).toBe(true); + + yield* Deferred.succeed(releaseConfig, PROBE_CAPABLE_CONFIG); + expect(yield* Deferred.await(remembered)).toBe(connectionRouteId(LAN_TARGET)); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("does not remember a sticky route when initial config fails", () => + Effect.gen(function* () { + const remembered = yield* Ref.make(false); + const harness = yield* makeHarness({ + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + initialConfig: () => Effect.fail(transient("Wrong environment.")), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_ENTRY, { + initiallyDesired: true, + rememberRoute: () => Ref.set(remembered, true).pipe(Effect.as(Option.none())), + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* TestClock.adjust("1 second"); + expect(yield* Ref.get(remembered)).toBe(false); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("waits while offline and connects immediately when the network returns", () => Effect.gen(function* () { const harness = yield* makeHarness({ networkStatus: "offline" }); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 1430cad92df2a..c7a028d7ffc78 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -1284,10 +1284,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const connectedAt = yield* Clock.currentTimeMillis; yield* setLease(connectedLease); if (connectedLease.activeRoute !== undefined && options?.rememberRoute !== undefined) { - // Persist sticky order for the next cold start only. Applying it to the - // live entry would make the active route look preferred and stop - // better-route upgrades (for example LAN while connected over relay). - yield* options.rememberRoute({ activeRoute: connectedLease.activeRoute }); + const activeRoute = connectedLease.activeRoute; + const remember = options.rememberRoute; + // Sticky for the next cold start only, and only after initialConfig. Too + // early prefers a route that later fails sync; applying it live would also + // stop better-route upgrades (LAN while on relay). + yield* connectedLease.session.initialConfig.pipe( + Effect.andThen(remember({ activeRoute })), + Effect.ignoreCause({ log: true }), + Effect.forkScoped, + ); } if (options?.learnRoutes !== undefined) { yield* learnRoutesFrom(options.learnRoutes, connectedLease).pipe(Effect.forkScoped); diff --git a/third-party-licenses.config.json b/third-party-licenses.config.json index 0ce031a30f359..79ac696d7f6a0 100644 --- a/third-party-licenses.config.json +++ b/third-party-licenses.config.json @@ -841,6 +841,26 @@ "name": "iobuffer", "license": "MIT", "sourceUrl": "https://github.com/image-js/iobuffer" + }, + { + "name": "deprecated-react-native-prop-types", + "version": "2.3.0", + "license": "MIT", + "sourceUrl": "https://github.com/facebook/react-native", + "generatedNotice": { + "licenseId": "MIT", + "copyrights": ["Copyright (c) Meta Platforms, Inc. and affiliates."] + } + }, + { + "name": "@react-native/normalize-color", + "version": "2.1.0", + "license": "MIT", + "sourceUrl": "https://github.com/facebook/react-native", + "generatedNotice": { + "licenseId": "MIT", + "copyrights": ["Copyright (c) Meta Platforms, Inc. and affiliates."] + } } ] }