From 88540690b5f4ca9fd93215b31db33ac17ee7fb8a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:37:39 +0000 Subject: [PATCH 1/3] test(client-runtime): a better route that answers keeps the live session until it connects Reproduces #24516: the supervisor releases a healthy session as soon as a higher-ranked route passes preflight, before that route has connected. AI-assisted; verified with the tests and tools listed above. --- .../src/connection/supervisor.test.ts | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 163075a6efa7f..943240e0600e5 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -2023,6 +2023,40 @@ describe("EnvironmentSupervisor routes", () => { }), ); + it.effect("keeps the live session until a better route that answered has connected", () => + Effect.gen(function* () { + const lanReachable = yield* Ref.make(false); + const releasesAtLanAttempt = 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" + ? Ref.get(harness.releaseCount).pipe( + Effect.flatMap((releases) => Deferred.succeed(releasesAtLanAttempt, releases)), + Effect.andThen(Effect.fail(transient("WebSocket upgrade failed."))), + ) + : 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, + ); + yield* Ref.set(lanReachable, true); + yield* TestClock.adjust("60 seconds"); + + expect(yield* Deferred.await(releasesAtLanAttempt)).toBe(0); + }), + ); + it.effect("keeps checking a LAN route every minute when it answers but cannot be prepared", () => Effect.gen(function* () { const lanPreflights = yield* Ref.make(0); From c67704cda123bca86e7e291668ea1787d742c540 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:44:57 +0000 Subject: [PATCH 2/3] fix(client-runtime): a better route connects before the live session is replaced The supervisor closed a healthy session as soon as a higher-ranked route answered its check, then fell back when that route could not open a socket, so remote connections flapped every one to five minutes. The better route now connects beside the live session, and the next attempt adopts that lease. A route that answers but cannot connect only earns its cooldown and never interrupts the session. Fixes pingdotgg/t3code#24516 AI-assisted; verified with the tests and tools listed above. --- .../src/connection/supervisor.test.ts | 2 +- .../src/connection/supervisor.ts | 215 +++++++++++------- 2 files changed, 133 insertions(+), 84 deletions(-) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 943240e0600e5..5da5e952991ea 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -2034,7 +2034,7 @@ describe("EnvironmentSupervisor routes", () => { Effect.map((reachable) => (reachable ? "answered" : "silent")), ) : Effect.succeed("unchecked"), - prepare: (_attempt, target) => + prepare: (_attempt, target): Effect.Effect => target._tag === "BearerConnectionTarget" ? Ref.get(harness.releaseCount).pipe( Effect.flatMap((releases) => Deferred.succeed(releasesAtLanAttempt, releases)), diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index bbeffd1cb7bc2..ca2660f555799 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -35,8 +35,8 @@ import { connectionRouteKind, connectionRoutes, type ConnectionRouteKind, - entryWithRoutes, insertRoute, + routeEntry, type ReportedEndpoint, } from "./routes.ts"; @@ -69,8 +69,8 @@ const BETTER_ROUTE_CHECK_INTERVAL_MS = 60_000; // that keeps failing is checked half as often each time, until the network // changes and it may answer again. const SILENT_ROUTE_BACKOFF_MAX_MS = 1_800_000; -// A better route that answered the check but then failed to connect is not -// tried again for this long, so a flaky LAN cannot bounce the connection. +// A better route that answered the check but then failed to connect beside +// the live session is not tried again for this long. const BETTER_ROUTE_COOLDOWN_MS = 300_000; interface SupervisorIntent { @@ -97,6 +97,11 @@ type ConnectedLease = ConnectionDriver.EnvironmentConnectionLease & { readonly activeRoute: ConnectionRoute | undefined; }; +interface ConnectedUpgrade { + readonly lease: ConnectionDriver.EnvironmentConnectionLease; + readonly scope: Scope.Closeable; +} + interface TracedAttemptFailure { readonly error: ConnectionAttemptError; readonly attemptSpan: Option.Option; @@ -319,7 +324,15 @@ 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()); + // A better route connects beside the live session, and the next attempt + // adopts it, so a route that answers its check but cannot connect never + // interrupts the session. + const upgrade = yield* Ref.make(Option.none()); + const dropUpgrade = Ref.getAndSet(upgrade, Option.none()).pipe( + Effect.flatMap((pending) => + Option.isSome(pending) ? Scope.close(pending.value.scope, Exit.void) : Effect.void, + ), + ); const routeCooldowns = yield* Ref.make>(new Map()); // The generation keeps a check that started before a reset from recording // its stale failures after it. @@ -332,19 +345,6 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( routes: new Map(), })); - const attemptEntry = Effect.gen(function* () { - const latest = yield* Ref.get(currentEntry); - const routes = connectionRoutes(latest); - const preferred = yield* Ref.getAndSet(preferredRouteId, Option.none()); - if (Option.isNone(preferred)) return latest; - const route = routes.find( - (candidate) => connectionRouteId(candidate.target) === preferred.value, - ); - return route === undefined - ? latest - : entryWithRoutes(latest, [route, ...routes.filter((candidate) => candidate !== route)]); - }); - const learnRoutesFrom = Effect.fnUntraced(function* ( learn: NonNullable, lease: ConnectedLease, @@ -472,7 +472,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( lastFailure: ConnectionAttemptError | null, synchronizedRoutes: Ref.Ref, ) { - return yield* driver.connect(yield* attemptEntry, (progress) => + return yield* driver.connect(yield* Ref.get(currentEntry), (progress) => (progress.stage === "synchronizing" ? Ref.update(synchronizedRoutes, (count) => count + 1) : Effect.void @@ -620,16 +620,47 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const isRelayLease = (lease: ConnectionDriver.EnvironmentConnectionLease) => lease.prepared.target._tag === "RelayConnectionTarget"; + const connectBetterRoute = (routeId: string) => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const latest = yield* Ref.get(currentEntry); + const route = connectionRoutes(latest).find( + (candidate) => connectionRouteId(candidate.target) === routeId, + ); + if (route === undefined) return; + const scope = yield* Scope.make(); + const connected = yield* restore( + driver + .connect(routeEntry(latest, route), () => Effect.void) + .pipe( + Scope.provide(scope), + Effect.timeoutOption(Duration.sum(establishmentTimeout, synchronizationTimeout)), + ), + ).pipe(Effect.exit); + if (Exit.isSuccess(connected) && Option.isSome(connected.value)) { + yield* Ref.set(upgrade, Option.some({ lease: connected.value.value, scope })); + return yield* signal({ _tag: "BetterRouteAvailable", routeId }); + } + yield* Scope.close(scope, Exit.void); + if (Exit.isFailure(connected) && Cause.hasInterruptsOnly(connected.cause)) return; + 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 connectedLeaseEnd = Effect.fnUntraced(function* ( next: SupervisorSignal, lease: ConnectionDriver.EnvironmentConnectionLease, + upgradeTo: (routeId: string) => Effect.Effect, ) { if (next._tag === "DisconnectRequested") { return "end" as const; } if (next._tag === "BetterRouteAvailable") { - yield* Ref.set(preferredRouteId, Option.some(next.routeId)); - return "reset" as const; + if (Option.isSome(yield* Ref.get(upgrade))) return "reset" as const; + yield* upgradeTo(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)) { @@ -757,10 +788,22 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Stream.runForEach(() => requestBetterRouteCheck(lease)), Effect.forkScoped, ); + let upgrading = false; + const upgradeTo = Effect.fnUntraced(function* (routeId: string) { + const better = (yield* currentRoutes).findIndex( + (route) => connectionRouteId(route.target) === routeId, + ); + if (upgrading || better === -1 || better >= (yield* routeIndex(lease))) return; + upgrading = true; + yield* connectBetterRoute(routeId).pipe( + Effect.ensuring(Effect.sync(() => (upgrading = false))), + Effect.forkScoped, + ); + }); const routeKind = yield* leaseRouteKind(lease); for (;;) { const next = yield* takeSignal; - const end = yield* connectedLeaseEnd(next, lease); + const end = yield* connectedLeaseEnd(next, lease, upgradeTo); if (end !== undefined) { return end === "reset"; } @@ -868,7 +911,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( yield* probeEvent.exit; break; } - const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease); + const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease, upgradeTo); if (endDuringProbe !== undefined) { yield* stopProbe; return endDuringProbe === "reset"; @@ -898,60 +941,72 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( pendingRetry: Option.Option, ignoreOffline: boolean, ) { - const switchingTo = yield* Ref.get(preferredRouteId); yield* SubscriptionRef.set(prepared, Option.none()); + const adopted = yield* Ref.getAndSet(upgrade, Option.none()); + if (Option.isSome(adopted)) { + yield* Effect.addFinalizer(() => Scope.close(adopted.value.scope, Exit.void)); + } const synchronizedRoutes = yield* Ref.make(0); - const establishment = yield* Effect.raceAllFirst([ - exitUnlessInterrupted( - establishTracedConnection( - attempt, - generation, - lastFailure, - pendingRetry, - synchronizedRoutes, - ), - ).pipe( - Effect.map((exit): EstablishmentEvent => ({ + const establishment: EstablishmentEvent = Option.isSome(adopted) + ? { _tag: "Completed", - exit, - })), - ), - waitForEstablishmentInterrupt().pipe( - Effect.map((resetRetry): EstablishmentEvent => ({ - _tag: "Interrupted", - 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 routes = connectionRoutes(yield* Ref.get(currentEntry)); - const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routes.length)); - // The desktop bounds SSH preparation, which may download a server - // release, so for SSH-only entries the clock stops while preparing. - const pausesWhilePreparing = routes.every((route) => connectionRouteKind(route) === "ssh"); - const preparing = (expected: boolean) => - SubscriptionRef.changes(state).pipe( - Stream.filter((current) => (current.stage === "preparing") === expected), - Stream.runHead, - ); - let spent = 0n; - while (true) { - if (pausesWhilePreparing) yield* preparing(false); - const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); - const remaining = - setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout) - spent; - if (remaining <= 0n) { - return { _tag: "TimedOut" } as const satisfies EstablishmentEvent; - } - const resumedAt = yield* Clock.monotonicTimeNanos; - const sleep = Effect.sleep(Duration.nanos(remaining)); - yield* pausesWhilePreparing ? Effect.raceFirst(sleep, preparing(true)) : sleep; - spent += (yield* Clock.monotonicTimeNanos) - resumedAt; + exit: Exit.succeed({ attemptSpan: Option.none(), lease: adopted.value.lease }), } - }), - ]); + : yield* Effect.raceAllFirst([ + exitUnlessInterrupted( + establishTracedConnection( + attempt, + generation, + lastFailure, + pendingRetry, + synchronizedRoutes, + ), + ).pipe( + Effect.map((exit): EstablishmentEvent => ({ + _tag: "Completed", + exit, + })), + ), + waitForEstablishmentInterrupt().pipe( + Effect.map((resetRetry): EstablishmentEvent => ({ + _tag: "Interrupted", + 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 routes = connectionRoutes(yield* Ref.get(currentEntry)); + const setup = Duration.toNanosUnsafe( + Duration.times(establishmentTimeout, routes.length), + ); + // The desktop bounds SSH preparation, which may download a server + // release, so for SSH-only entries the clock stops while preparing. + const pausesWhilePreparing = routes.every( + (route) => connectionRouteKind(route) === "ssh", + ); + const preparing = (expected: boolean) => + SubscriptionRef.changes(state).pipe( + Stream.filter((current) => (current.stage === "preparing") === expected), + Stream.runHead, + ); + let spent = 0n; + while (true) { + if (pausesWhilePreparing) yield* preparing(false); + const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); + const remaining = + setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout) - spent; + if (remaining <= 0n) { + return { _tag: "TimedOut" } as const satisfies EstablishmentEvent; + } + const resumedAt = yield* Clock.monotonicTimeNanos; + const sleep = Effect.sleep(Duration.nanos(remaining)); + yield* pausesWhilePreparing ? Effect.raceFirst(sleep, preparing(true)) : sleep; + spent += (yield* Clock.monotonicTimeNanos) - resumedAt; + } + }), + ]); if (establishment._tag === "Interrupted") { return { @@ -995,16 +1050,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 { @@ -1138,12 +1183,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( resetRetryLadder(); latestFailure = null; yield* clearLease; + yield* dropUpgrade; yield* setState(availableState(currentIntent, generation)); yield* waitForSignal; continue; } if (currentIntent.network === "offline" && !replacing) { yield* clearLease; + yield* dropUpgrade; yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure)); const applicationActivated = yield* waitForSignal; if (applicationActivated) { @@ -1296,7 +1343,9 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Effect.withSpan("EnvironmentSupervisor.retryNow"), ); - yield* Effect.addFinalizer(() => Queue.shutdown(signals).pipe(Effect.andThen(clearLease))); + yield* Effect.addFinalizer(() => + Queue.shutdown(signals).pipe(Effect.andThen(clearLease), Effect.andThen(dropUpgrade)), + ); return EnvironmentSupervisor.of({ target, From 0c319544a9fdea86f2a8ac6d05c72e4ea1444b91 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:57:49 +0000 Subject: [PATCH 3/3] fix(client-runtime): a learned route connects with the routes whose token it borrows Connecting a better route beside the live session passed only that route, so a learned plain HTTP route that borrows another route's token was no longer usable and never connected. Pass the whole entry with the better route first, and adopt the lease only when it landed on that route. AI-assisted; verified with the tests and tools listed above. --- .../src/connection/supervisor.test.ts | 71 +++++++++++++++++++ .../src/connection/supervisor.ts | 20 ++++-- 2 files changed, 85 insertions(+), 6 deletions(-) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 5da5e952991ea..6367737ff3480 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -2057,6 +2057,77 @@ describe("EnvironmentSupervisor routes", () => { }), ); + it.effect("moves to a learned LAN route that borrows a plain HTTP pairing's token", () => + Effect.gen(function* () { + const publicTarget = new BearerConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + connectionId: "bearer:public", + }); + const learnedLanTarget = new BearerConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + connectionId: `learned:${TARGET.environmentId}:http://192.168.1.10:3773@bearer:public`, + }); + const entry: ConnectionCatalogEntry = { + target: learnedLanTarget, + profile: Option.some( + new BearerConnectionProfile({ + connectionId: learnedLanTarget.connectionId, + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: "http://192.168.1.10:3773/", + wsBaseUrl: "ws://192.168.1.10:3773/", + learned: true, + }), + ), + alternateRoutes: [ + { + target: publicTarget, + profile: Option.some( + new BearerConnectionProfile({ + connectionId: publicTarget.connectionId, + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: "http://203.0.113.5:3773/", + wsBaseUrl: "ws://203.0.113.5:3773/", + }), + ), + }, + ], + enabled: true, + }; + const lanReachable = yield* Ref.make(false); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target === learnedLanTarget + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("answered"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(entry, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + yield* Ref.set(lanReachable, true); + yield* TestClock.adjust("60 seconds"); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target).toBe( + learnedLanTarget, + ); + }), + ); + it.effect("keeps checking a LAN route every minute when it answers but cannot be prepared", () => Effect.gen(function* () { const lanPreflights = yield* Ref.make(0); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index ca2660f555799..75d924c1ac871 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -35,8 +35,8 @@ import { connectionRouteKind, connectionRoutes, type ConnectionRouteKind, + entryWithRoutes, insertRoute, - routeEntry, type ReportedEndpoint, } from "./routes.ts"; @@ -624,20 +624,28 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const latest = yield* Ref.get(currentEntry); - const route = connectionRoutes(latest).find( - (candidate) => connectionRouteId(candidate.target) === routeId, - ); + const routes = connectionRoutes(latest); + const route = routes.find((candidate) => connectionRouteId(candidate.target) === routeId); if (route === undefined) return; const scope = yield* Scope.make(); + // The whole entry goes along, because a learned route may borrow the + // token of another route, and only a lease over this route counts. const connected = yield* restore( driver - .connect(routeEntry(latest, route), () => Effect.void) + .connect( + entryWithRoutes(latest, [route, ...routes.filter((other) => other !== route)]), + () => Effect.void, + ) .pipe( Scope.provide(scope), Effect.timeoutOption(Duration.sum(establishmentTimeout, synchronizationTimeout)), ), ).pipe(Effect.exit); - if (Exit.isSuccess(connected) && Option.isSome(connected.value)) { + if ( + Exit.isSuccess(connected) && + Option.isSome(connected.value) && + connectionRouteId(connected.value.value.prepared.target) === routeId + ) { yield* Ref.set(upgrade, Option.some({ lease: connected.value.value, scope })); return yield* signal({ _tag: "BetterRouteAvailable", routeId }); }