diff --git a/docs/internals/remote.md b/docs/internals/remote.md index 4a348382ac47b..18674a6abd1c5 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -37,7 +37,13 @@ network receives no credential. That check is not proof of a working route: when every route stays silent, each is still tried. A route that fails to connect, including a blocked one such as a signed-out T3 Connect, moves on to the next; only an incompatible server stops the walk, because it is the same -server on every route. While connected over a later route the +server on every route. A direct route that answered the check but is still +connecting when the check's head start ends starts T3 Connect instead of holding +it back. T3 Connect then races that route and every direct route after it, so a +failed direct route still moves on to the next one. The first to connect wins and +the others close. An incompatible T3 Connect server does not end that race, because +relay can reach a different server than a direct route, such as a preview app +sharing the home directory; it is reported only when no direct route connects. While connected over a later route the [supervisor](../../packages/client-runtime/src/connection/supervisor.ts) preflights the earlier ones and replaces the session when one would connect. Preflight includes authorization so a route that answers but rejects this diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index 416ca526db7a7..e0509af2a841c 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -4,6 +4,7 @@ import * as Option from "effect/Option"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import type * 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"; @@ -85,8 +86,10 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") let transient: ConnectionAttemptError | undefined; let blocked: ConnectionAttemptError | undefined; // Each route gets its own scope so a half-open session closes before the next try. - const attempt = Effect.fnUntraced(function* (route: ConnectionRoute) { - const routeScope = yield* Scope.fork(attemptScope); + const attempt = Effect.fnUntraced(function* ( + route: ConnectionRoute, + routeScope: Scope.Closeable, + ) { const result = yield* connectRoute(route).pipe( Scope.provide(routeScope), Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : Scope.close(routeScope, exit))), @@ -98,8 +101,64 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") } return result; }); + const relayIndex = routes.findIndex((route) => route.target._tag === "RelayConnectionTarget"); + const relayRoute = routes[relayIndex]; + type Settled = { + readonly result: Result.Result; + readonly scope: Scope.Closeable; + }; + let relay: + | { + readonly fiber: Fiber.Fiber< + Settled, + Result.Result + >; + readonly scope: Scope.Closeable; + } + | undefined; + // A direct route still connecting when the head start ends starts relay, which + // then races every direct attempt after it. The first to connect wins. A failed + // direct route moves on to the next one while relay keeps connecting. + const attemptRacingRelay = Effect.fnUntraced(function* (route: ConnectionRoute) { + const directScope = yield* Scope.fork(attemptScope); + const direct = yield* Effect.forkChild(attempt(route, directScope)); + if (relay === undefined) { + const early = yield* Fiber.join(direct).pipe(Effect.timeoutOption(ROUTE_CHECK_HEAD_START_MS)); + if (Option.isSome(early)) return early.value; + const relayScope = yield* Scope.fork(attemptScope); + const fiber = yield* Effect.forkChild( + attempt(relayRoute!, relayScope).pipe( + Effect.flatMap((result) => + // Relay can reach a different server than the direct routes, so an + // incompatible relay is a failed backup, not the end of the walk. + result._tag === "Success" + ? Effect.succeed({ result, scope: relayScope }) + : Effect.fail(result), + ), + ), + ); + relay = { fiber, scope: relayScope }; + } + const racing = relay; + const raced = yield* Effect.race( + Fiber.join(direct).pipe(Effect.map((result): Settled => ({ result, scope: directScope }))), + // A failed relay leaves the direct attempt to finish on its own. + Fiber.join(racing.fiber).pipe(Effect.catch(() => Effect.never)), + ); + if (raced.scope === directScope) { + if (raced.result._tag === "Success" || raced.result.failure.reason === "unsupported") { + yield* Fiber.interrupt(racing.fiber); + yield* Scope.close(racing.scope, Exit.void); + } + } else { + yield* Fiber.interrupt(direct); + yield* Scope.close(directScope, Exit.void); + } + return raced.result; + }); const silent: Array<{ readonly route: ConnectionRoute; readonly index: number }> = []; for (const [index, route] of routes.entries()) { + if (index === relayIndex && relay !== undefined) continue; const check = checks.length === 0 ? "unchecked" @@ -111,11 +170,21 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") silent.push({ route, index }); continue; } - const result = yield* attempt(route); + const result = + relay !== undefined || + (check === "answered" && relayRoute !== undefined && index < relayIndex) + ? yield* attemptRacingRelay(route) + : yield* attempt(route, yield* Scope.fork(attemptScope)); 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; } + if (relay !== undefined) { + const settled = yield* Fiber.join(relay.fiber).pipe(Effect.result); + const result = settled._tag === "Success" ? settled.success.result : settled.failure; + if (result._tag === "Success") return result.success; + if (result.failure.reason === "unsupported") return yield* result.failure; + } const late = yield* Effect.forEach(silent, ({ index }) => Fiber.join(checks[index]!), { concurrency: "unbounded", }); @@ -126,7 +195,7 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") // longer than the check did. Once another route failed in a way that // retries, the retry reaches it sooner than waiting here would. if (transient !== undefined && stillSilent.some((entry) => entry.route === route)) break; - const result = yield* attempt(route); + const result = yield* attempt(route, yield* Scope.fork(attemptScope)); if (result._tag === "Success") return result.success; if (result.failure.reason === "unsupported") return yield* result.failure; } diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 1fb7aaed524a8..3f2ae32608121 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -45,6 +45,7 @@ import { type SupervisorConnectionState, } from "./model.ts"; import * as RpcSession from "../rpc/session.ts"; +import { environmentMismatchError } from "./errors.ts"; import { presentConnectionState } from "./presentation.ts"; import * as EnvironmentSupervisor from "./supervisor.ts"; import * as ConnectionWakeups from "./wakeups.ts"; @@ -109,6 +110,30 @@ const LAN_THEN_RELAY_ENTRY: ConnectionCatalogEntry = { enabled: true, }; +const SECOND_LAN_TARGET = new BearerConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + connectionId: "bearer:second-lan", +}); +const TWO_LANS_THEN_RELAY_ENTRY: ConnectionCatalogEntry = { + ...LAN_THEN_RELAY_ENTRY, + alternateRoutes: [ + { + target: SECOND_LAN_TARGET, + profile: Option.some( + new BearerConnectionProfile({ + connectionId: SECOND_LAN_TARGET.connectionId, + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: "http://192.168.56.1:3773/", + wsBaseUrl: "ws://192.168.56.1:3773/", + }), + ), + }, + { target: RELAY_TARGET, profile: Option.none() }, + ], +}; + const SSH_ENTRY: ConnectionCatalogEntry = { target: new SshConnectionTarget({ environmentId: TARGET.environmentId, @@ -1841,29 +1866,6 @@ describe("EnvironmentSupervisor routes", () => { it.effect("shares one head start across several slow LAN routes", () => Effect.gen(function* () { - const secondLanTarget = new BearerConnectionTarget({ - environmentId: TARGET.environmentId, - label: TARGET.label, - connectionId: "bearer:second-lan", - }); - const entry: ConnectionCatalogEntry = { - ...LAN_THEN_RELAY_ENTRY, - alternateRoutes: [ - { - target: secondLanTarget, - profile: Option.some( - new BearerConnectionProfile({ - connectionId: secondLanTarget.connectionId, - environmentId: TARGET.environmentId, - label: TARGET.label, - httpBaseUrl: "http://192.168.56.1:3773/", - wsBaseUrl: "ws://192.168.56.1:3773/", - }), - ), - }, - { target: RELAY_TARGET, profile: Option.none() }, - ], - }; const checking = yield* Ref.make(0); const checkingLans = yield* Deferred.make(); const harness = yield* makeHarness({ @@ -1879,7 +1881,7 @@ describe("EnvironmentSupervisor routes", () => { : Effect.succeed("unchecked"), prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), }); - const supervisor = yield* EnvironmentSupervisor.make(entry, { + const supervisor = yield* EnvironmentSupervisor.make(TWO_LANS_THEN_RELAY_ENTRY, { initiallyDesired: true, }).pipe(Effect.provide(harness.dependencies)); @@ -1919,6 +1921,216 @@ describe("EnvironmentSupervisor routes", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect.each(["hangs", "fails late"] as const)( + "connects over relay when an answering LAN route's readiness %s", + (lanReadiness) => + Effect.gen(function* () { + const lanWaiting = yield* Deferred.make(); + const relayMayConnect = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + Effect.succeed( + route.target._tag === "BearerConnectionTarget" ? "answered" : "unchecked", + ), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + ready: (attempt) => + attempt === 1 + ? Deferred.succeed(lanWaiting, undefined).pipe( + Effect.andThen( + lanReadiness === "hangs" + ? Effect.never + : Effect.sleep("5 seconds").pipe( + Effect.andThen(Effect.fail(transient("LAN readiness failed."))), + ), + ), + ) + : Deferred.await(relayMayConnect), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(lanWaiting); + yield* TestClock.adjust("500 millis"); + // A late LAN failure does not end the race while relay is still connecting. + yield* TestClock.adjust("5 seconds"); + yield* Deferred.succeed(relayMayConnect, undefined); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "RelayConnectionTarget", + ); + // The stuck LAN session is closed rather than left beside the relay one. + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("gives a later answering route its own head start before racing relay", () => + Effect.gen(function* () { + const isFirstLan = (target: ConnectionTarget) => + connectionRouteId(target) === LAN_TARGET.connectionId; + const firstLanPreparing = yield* Deferred.make(); + const secondLanWaiting = yield* Deferred.make(); + const attempted: string[] = []; + const harness = yield* makeHarness({ + checkRoute: (route) => + Effect.succeed( + isFirstLan(route.target) || route.target._tag !== "BearerConnectionTarget" + ? "unchecked" + : "answered", + ), + // The first LAN route spends the whole head start failing. + prepare: (_attempt, target) => { + attempted.push(connectionRouteId(target)); + return isFirstLan(target) + ? Deferred.succeed(firstLanPreparing, undefined).pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen(Effect.fail(transient("First LAN route failed."))), + ) + : Effect.succeed(preparedFor(target)); + }, + ready: () => + Deferred.succeed(secondLanWaiting, undefined).pipe( + Effect.andThen(Effect.sleep("200 millis")), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(TWO_LANS_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(firstLanPreparing); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(secondLanWaiting); + yield* TestClock.adjust("200 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect( + connectionRouteId( + Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target, + ), + ).toBe(SECOND_LAN_TARGET.connectionId); + expect(attempted).toEqual([LAN_TARGET.connectionId, SECOND_LAN_TARGET.connectionId]); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("tries the next answering LAN route when the raced one fails while relay hangs", () => + Effect.gen(function* () { + const firstLanWaiting = yield* Deferred.make(); + const relayWaiting = yield* Deferred.make(); + const attempted: string[] = []; + const harness = yield* makeHarness({ + checkRoute: (route) => + Effect.succeed(route.target._tag === "BearerConnectionTarget" ? "answered" : "unchecked"), + prepare: (_attempt, target) => { + attempted.push(connectionRouteId(target)); + return Effect.succeed(preparedFor(target)); + }, + ready: (attempt) => + attempt === 1 + ? Deferred.succeed(firstLanWaiting, undefined).pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen(Effect.fail(transient("First LAN readiness failed."))), + ) + : attempt === 2 + ? Deferred.succeed(relayWaiting, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + }); + const supervisor = yield* EnvironmentSupervisor.make(TWO_LANS_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(firstLanWaiting); + yield* TestClock.adjust("500 millis"); + yield* Deferred.await(relayWaiting); + yield* TestClock.adjust("500 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect( + connectionRouteId( + Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target, + ), + ).toBe(SECOND_LAN_TARGET.connectionId); + expect(attempted).toEqual([ + LAN_TARGET.connectionId, + connectionRouteId(RELAY_TARGET), + SECOND_LAN_TARGET.connectionId, + ]); + // The failed LAN session and the still-connecting relay session both close. + expect(yield* Ref.get(harness.sessionCount)).toBe(3); + expect(yield* Ref.get(harness.releaseCount)).toBe(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect.each(["fails its identity check", "reaches an unsupported server"] as const)( + "keeps waiting for a slow LAN route when relay %s", + (relayFailure) => + Effect.gen(function* () { + const lanWaiting = yield* Deferred.make(); + const attempted: string[] = []; + const harness = yield* makeHarness({ + checkRoute: (route) => + Effect.succeed( + route.target._tag === "BearerConnectionTarget" ? "answered" : "unchecked", + ), + prepare: (_attempt, target) => { + attempted.push(target._tag); + return target._tag === "RelayConnectionTarget" + ? Effect.fail( + relayFailure === "fails its identity check" + ? environmentMismatchError({ + expected: TARGET.environmentId, + actual: EnvironmentId.make("environment-2"), + }) + : new ConnectionBlockedError({ + reason: "unsupported", + detail: "This client requires a newer server.", + }), + ) + : Effect.succeed(preparedFor(target)); + }, + ready: () => + Deferred.succeed(lanWaiting, undefined).pipe(Effect.andThen(Effect.sleep("3 seconds"))), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(lanWaiting); + yield* TestClock.adjust("3 seconds"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "BearerConnectionTarget", + ); + expect(attempted).toEqual(["BearerConnectionTarget", "RelayConnectionTarget"]); + expect(yield* Ref.get(harness.releaseCount)).toBe(0); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("closes both racing routes when the user disconnects", () => + Effect.gen(function* () { + const lanWaiting = yield* Deferred.make(); + const relayWaiting = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + Effect.succeed(route.target._tag === "BearerConnectionTarget" ? "answered" : "unchecked"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + ready: (attempt) => + Deferred.succeed(attempt === 1 ? lanWaiting : relayWaiting, undefined).pipe( + Effect.andThen(Effect.never), + ), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(lanWaiting); + yield* TestClock.adjust("500 millis"); + yield* Deferred.await(relayWaiting); + 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("tries a slow tailnet route that answered before a dead LAN route", () => Effect.gen(function* () { const tailnetTarget = new BearerConnectionTarget({