From f36df1852423fef7f7855f86946c9f9a95c3429f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 5 Oct 2026 19:38:13 -0700 Subject: [PATCH] fix(web): a rejected desktop-local credential is not retried every poll (#16273) Co-authored-by: Claude Opus 5.5 (1M context) --- apps/web/src/connection/platform.test.ts | 70 +++++++++++ apps/web/src/connection/platform.ts | 117 +++++++++++++++++- .../src/connection/registry.test.ts | 111 +++++++++++++++++ .../client-runtime/src/connection/registry.ts | 49 ++++++-- 4 files changed, 334 insertions(+), 13 deletions(-) diff --git a/apps/web/src/connection/platform.test.ts b/apps/web/src/connection/platform.test.ts index ee45e2932e3910ad66a9d6298002b3e834b0aa06..9cd1a807da79649d6d7578a2ccc6f73c85079e04 100644 --- a/apps/web/src/connection/platform.test.ts +++ b/apps/web/src/connection/platform.test.ts @@ -7,10 +7,17 @@ import { } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import { + ConnectionBlockedError, + ConnectionTransientError, +} from "@t3tools/client-runtime/connection"; import { canRetainCachedPlatformRegistrationAfterRefreshFailure, canReuseCachedPlatformRegistration, + isRejectedBootstrapCredentialError, + isRejectedSecondaryBootstrap, + nextRejectedSecondaryBootstrap, primaryRegistrationToRetainAfterTopologyRead, provisionDesktopSshEnvironment, readPrimaryEnvironmentTargetResult, @@ -202,6 +209,69 @@ describe("desktop-local bearer cache", () => { }); }); +describe("rejected desktop-local bootstrap tokens", () => { + it("backs off on a rejected signature and retries it on a growing, capped delay", () => { + const signature = "http://a|ws://a|old-token"; + const first = nextRejectedSecondaryBootstrap(undefined, signature, 0); + + expect(isRejectedSecondaryBootstrap(first, signature, 59_999)).toBe(true); + // A backend restarted on the same port accepts the same token again, so it is retried. + expect(isRejectedSecondaryBootstrap(first, signature, 60_000)).toBe(false); + + const second = nextRejectedSecondaryBootstrap(first, signature, 60_000); + expect(second.retryAtEpochMs).toBe(180_000); + + let backoff = second; + for (let attempt = 0; attempt < 10; attempt += 1) { + backoff = nextRejectedSecondaryBootstrap(backoff, signature, 0); + } + expect(backoff.delayMs).toBe(30 * 60_000); + }); + + it("retries at once when the token or endpoint changes", () => { + const rejected = nextRejectedSecondaryBootstrap(undefined, "http://a|ws://a|old-token", 0); + + expect(isRejectedSecondaryBootstrap(rejected, "http://a|ws://a|new-token", 1)).toBe(false); + expect(isRejectedSecondaryBootstrap(rejected, "http://b|ws://b|old-token", 1)).toBe(false); + expect(isRejectedSecondaryBootstrap(undefined, "http://a|ws://a|old-token", 1)).toBe(false); + expect(nextRejectedSecondaryBootstrap(rejected, "http://a|ws://a|new-token", 1).delayMs).toBe( + 60_000, + ); + }); + + it("treats only authentication rejections as a dead credential", () => { + expect( + isRejectedBootstrapCredentialError( + new ConnectionBlockedError({ reason: "authentication", detail: "invalid" }), + ), + ).toBe(true); + expect( + isRejectedBootstrapCredentialError( + new ConnectionTransientError({ reason: "endpoint-unavailable", detail: "booting" }), + ), + ).toBe(false); + }); +}); + +describe("desktop-local bearer cache across token changes", () => { + it("keeps a live bearer when only the bootstrap token rotates", () => { + // Cached by endpoint: a rotated token reuses the bearer instead of + // dropping the environment (and its drafts) when the new token is rejected. + const cached = { + signature: "http://a|ws://a", + registration: {} as never, + expiresAtEpochMs: 20_000, + refreshAtEpochMs: 15_000, + }; + + expect(canReuseCachedPlatformRegistration(cached, "http://a|ws://a", 10_000)).toBe(true); + expect( + canRetainCachedPlatformRegistrationAfterRefreshFailure(cached, "http://a|ws://a", 16_000), + ).toBe(true); + expect(canReuseCachedPlatformRegistration(cached, "http://b|ws://b", 10_000)).toBe(false); + }); +}); + describe("primary topology cache", () => { const registration = {} as never; const cached = { diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index 1703402176018e9ce49966ba59b6d795aa221c2a..5352876f214cc331e01912b5e46bdb16e9f8cc50 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -9,6 +9,7 @@ import { BearerConnectionRegistration, BearerConnectionTarget, ConnectionBlockedError, + type ConnectionAttemptError, ConnectionTransientError, Connectivity, mapRemoteEnvironmentError, @@ -420,7 +421,7 @@ const loadSecondaryConnectionRegistration = Effect.fn( // Poll cadence for the desktop bootstrap topology. There is no change event on // the bridge, so the renderer polls; successful registrations are cached by a -// signature of their endpoint + token until bearer credentials approach expiry. +// signature of their endpoint until bearer credentials approach expiry. const PLATFORM_POLL_INTERVAL = "3 seconds"; const SECONDARY_BEARER_REFRESH_SKEW_MS = 5_000; @@ -447,6 +448,8 @@ interface CachedPlatformRegistration { readonly registration: PlatformConnectionRegistration; readonly expiresAtEpochMs?: number; readonly refreshAtEpochMs?: number; + /** The bootstrap token a desktop-local bearer was exchanged for. */ + readonly bootstrapToken?: string; } export type PrimaryEnvironmentTargetRead = @@ -499,6 +502,52 @@ export function canRetainCachedPlatformRegistrationAfterRefreshFailure( ); } +const REJECTED_BOOTSTRAP_RETRY_INITIAL_MS = 60_000; +const REJECTED_BOOTSTRAP_RETRY_MAX_MS = 30 * 60_000; + +/** A bootstrap token a backend rejected, and when the poll may try it again. */ +export interface RejectedSecondaryBootstrap { + readonly signature: string; + readonly retryAtEpochMs: number; + readonly delayMs: number; +} + +/** + * A backend that rejected a bootstrap token will usually keep rejecting it, so + * the poll backs off on that exact signature instead of re-presenting a dead + * credential every few seconds. It still retries on a capped backoff: a + * backend that restarts on the same port seeds a fresh grant for the same + * token, and nothing in the topology says it restarted. A new token or + * endpoint retries at once. + */ +export function isRejectedSecondaryBootstrap( + rejected: RejectedSecondaryBootstrap | undefined, + signature: string, + nowEpochMs: number, +): rejected is RejectedSecondaryBootstrap { + return ( + rejected !== undefined && + rejected.signature === signature && + nowEpochMs < rejected.retryAtEpochMs + ); +} + +export function nextRejectedSecondaryBootstrap( + previous: RejectedSecondaryBootstrap | undefined, + signature: string, + nowEpochMs: number, +): RejectedSecondaryBootstrap { + const delayMs = + previous?.signature === signature + ? Math.min(previous.delayMs * 2, REJECTED_BOOTSTRAP_RETRY_MAX_MS) + : REJECTED_BOOTSTRAP_RETRY_INITIAL_MS; + return { signature, retryAtEpochMs: nowEpochMs + delayMs, delayMs }; +} + +export function isRejectedBootstrapCredentialError(error: ConnectionAttemptError): boolean { + return error._tag === "ConnectionBlockedError" && error.reason === "authentication"; +} + export function secondaryRegistrationsToRetainAfterTopologyRead( previous: ReadonlyMap, topologyRead: DesktopSecondaryBootstrapsRead, @@ -523,6 +572,7 @@ const layerPlatformConnectionSource = Layer.effect( }); } const cacheRef = yield* Ref.make(new Map()); + const rejectedRef = yield* Ref.make(new Map()); // Resolve the full set of platform-managed environments the host currently // reports: the primary (same-origin cookie auth) plus any desktop-local @@ -588,17 +638,45 @@ const layerPlatformConnectionSource = Layer.effect( cause: topologyRead.cause, }); } else { + const rejected = yield* Ref.get(rejectedRef); + const nextRejected = new Map(); for (const bootstrap of topologyRead.bootstraps) { - const signature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}|${bootstrap.bootstrapToken ?? ""}`; + // The cached bearer belongs to the endpoint, not to the bootstrap + // token it was exchanged for: a new token must not drop a live + // session (its removal also clears the environment's drafts). The + // token only decides whether a rejected exchange is retried. + const endpointSignature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}`; + const signature = `${endpointSignature}|${bootstrap.bootstrapToken ?? ""}`; const cached = previous.get(bootstrap.id); + // A new token from the desktop means something changed (rotation, a + // restarted backend): exchange it rather than reusing a bearer that + // may be dead. The old bearer stays registered if that exchange fails. if ( cached !== undefined && - canReuseCachedPlatformRegistration(cached, signature, nowEpochMs) + cached.bootstrapToken === bootstrap.bootstrapToken && + canReuseCachedPlatformRegistration(cached, endpointSignature, nowEpochMs) ) { next.set(bootstrap.id, cached); registrations.push(cached.registration); continue; } + const previouslyRejected = rejected.get(bootstrap.id); + if (isRejectedSecondaryBootstrap(previouslyRejected, signature, nowEpochMs)) { + nextRejected.set(bootstrap.id, previouslyRejected); + // The bearer minted before the token died is still good until it expires. + if ( + cached !== undefined && + canRetainCachedPlatformRegistrationAfterRefreshFailure( + cached, + endpointSignature, + nowEpochMs, + ) + ) { + next.set(bootstrap.id, cached); + registrations.push(cached.registration); + } + continue; + } const built = yield* loadSecondaryConnectionRegistration(bootstrap).pipe( Effect.tapError((error) => Effect.logWarning("Could not connect a desktop-local backend.", { @@ -606,20 +684,49 @@ const layerPlatformConnectionSource = Layer.effect( error, }), ), + Effect.tapError((error) => + isRejectedBootstrapCredentialError(error) + ? // Back off from when the rejection arrived; the exchanges in + // this poll can take longer than the first backoff step. + Clock.currentTimeMillis.pipe( + Effect.map((rejectedAtEpochMs) => + nextRejected.set( + bootstrap.id, + nextRejectedSecondaryBootstrap( + previouslyRejected, + signature, + rejectedAtEpochMs, + ), + ), + ), + ) + : Effect.void, + ), Effect.option, ); if (Option.isSome(built)) { - const cacheEntry = { signature, ...built.value }; + const cacheEntry = { + signature: endpointSignature, + ...(bootstrap.bootstrapToken === undefined + ? {} + : { bootstrapToken: bootstrap.bootstrapToken }), + ...built.value, + }; next.set(bootstrap.id, cacheEntry); registrations.push(built.value.registration); } else if ( cached !== undefined && - canRetainCachedPlatformRegistrationAfterRefreshFailure(cached, signature, nowEpochMs) + canRetainCachedPlatformRegistrationAfterRefreshFailure( + cached, + endpointSignature, + nowEpochMs, + ) ) { next.set(bootstrap.id, cached); registrations.push(cached.registration); } } + yield* Ref.set(rejectedRef, nextRejected); } yield* Ref.set(cacheRef, next); diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 0d249ffab4060346b9d565a89f6a489bb6bb3e56..12296df81a3079e6b545bf7008570415c5fb05cc 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -158,6 +158,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( ) => Effect.Effect; readonly checkRoute?: (route: ConnectionRoute) => RouteCheck; readonly prepareRoute?: (target: ConnectionTarget) => ConnectionBlockedError | undefined; + /** Runs before preparing a route; may suspend to hold an attempt in flight. */ + readonly beforePrepareRoute?: ( + target: ConnectionTarget, + ) => Effect.Effect; readonly initialDisabled?: ReadonlyArray; }, ) { @@ -385,6 +389,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( target, }; if (options?.prepareError) return yield* options.prepareError; + const heldError = options?.beforePrepareRoute + ? yield* options.beforePrepareRoute(target) + : undefined; + if (heldError !== undefined) return yield* heldError; const routeError = options?.prepareRoute?.(target); if (routeError !== undefined) return yield* routeError; yield* Ref.update(connectedRoutes, (current) => [...current, connectionRouteId(target)]); @@ -1270,6 +1278,109 @@ describe("EnvironmentRegistry", () => { }), ); + it.effect("a replaced platform bearer wakes a supervisor blocked on the old one", () => + Effect.gen(function* () { + let storedCredentials: Ref.Ref> | undefined; + const harness = yield* makeHarness([], [], [], { + // Authentication fails while the old bearer is stored, as the server + // would answer once that session was revoked. + beforePrepareRoute: (target) => + target._tag !== "BearerConnectionTarget" || storedCredentials === undefined + ? Effect.succeed(undefined) + : Ref.get(storedCredentials).pipe( + Effect.map((current) => + current.get(target.connectionId)?.token === "old-bearer" + ? new ConnectionBlockedError({ reason: "authentication", detail: "revoked" }) + : undefined, + ), + ), + }); + storedCredentials = harness.storedCredentials; + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const registration = (token: string) => + new BearerConnectionRegistration({ + target: BEARER_TARGET, + profile: BEARER_PROFILE, + credential: new BearerConnectionCredential({ token }), + }); + yield* registry.reconcilePlatform([registration("old-bearer")]); + yield* awaitConnectionState( + registry, + BEARER_TARGET.environmentId, + (state) => state.phase === "blocked", + ); + + yield* registry.reconcilePlatform([registration("new-bearer")]); + + yield* awaitConnectionState( + registry, + BEARER_TARGET.environmentId, + (state) => state.phase === "connected", + ); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + + it.effect("a bearer replaced while an attempt is in flight retries with the new one", () => + Effect.gen(function* () { + const attemptStarted = yield* Deferred.make(); + const releaseFirstAttempt = yield* Deferred.make(); + const attempts = yield* Ref.make(0); + const harness = yield* makeHarness([], [], [], { + // The first attempt read the old bearer; it stalls, then fails + // authentication after the bearer has already been replaced. + beforePrepareRoute: (target) => + target._tag !== "BearerConnectionTarget" + ? Effect.succeed(undefined) + : Ref.updateAndGet(attempts, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 + ? Deferred.succeed(attemptStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirstAttempt)), + Effect.as( + new ConnectionBlockedError({ + reason: "authentication", + detail: "revoked", + }), + ), + ) + : Effect.succeed(undefined), + ), + ), + }); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const registration = (token: string) => + new BearerConnectionRegistration({ + target: BEARER_TARGET, + profile: BEARER_PROFILE, + credential: new BearerConnectionCredential({ token }), + }); + yield* registry.reconcilePlatform([registration("old-bearer")]); + yield* Deferred.await(attemptStarted).pipe(Effect.timeout("1 second")); + + // A live state subscription, as the UI holds, keeps the runtime in place + // so the stalled attempt's failure lands on the same supervisor. + yield* registry + .stateChanges(BEARER_TARGET.environmentId) + .pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })); + yield* Effect.yieldNow; + yield* registry.reconcilePlatform([registration("new-bearer")]); + yield* Deferred.succeed(releaseFirstAttempt, undefined); + + yield* awaitConnectionState( + registry, + BEARER_TARGET.environmentId, + (state) => state.phase === "connected", + ); + expect(yield* Ref.get(attempts)).toBeGreaterThanOrEqual(2); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("starts a newly paired bearer environment without re-reading its profile", () => 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 e0c45445e25f6c56556c0f90639d02e6996d33c6..d6ffac55c3dbeb23092f71554fa90df2e1143dd6 100644 --- a/packages/client-runtime/src/connection/registry.ts +++ b/packages/client-runtime/src/connection/registry.ts @@ -15,6 +15,7 @@ import * as SubscriptionRef from "effect/SubscriptionRef"; import * as ClientCapabilities from "../platform/capabilities.ts"; import { type ConnectionCatalogEntry, + type ConnectionCredential, type ConnectionProfile, type ConnectionRegistration, type ConnectionRoute, @@ -654,15 +655,26 @@ export const make = Effect.gen(function* () { // on their own loopback origin, so they authenticate with a bearer // token instead of the primary's same-origin cookie. Stash it where // the resolver's bearer broker looks it up. + let bearerReplaced = false; if (registration._tag === "BearerConnectionRegistration") { - yield* credentials.put(registration.target.connectionId, registration.credential).pipe( - Effect.catch((error) => - Effect.logWarning("Could not store the platform bearer credential.", { - environmentId: target.environmentId, - error, - }), - ), - ); + const stored = yield* credentials + .get(registration.target.connectionId) + .pipe(Effect.orElseSucceed(() => Option.none())); + const changed = + Option.isSome(stored) && !Equal.equals(stored.value, registration.credential); + // Only a bearer that was actually stored can be retried with; a + // failed write leaves the rejected one in place. + bearerReplaced = yield* credentials + .put(registration.target.connectionId, registration.credential) + .pipe( + Effect.as(changed), + Effect.catch((error) => + Effect.logWarning("Could not store the platform bearer credential.", { + environmentId: target.environmentId, + error, + }).pipe(Effect.as(false)), + ), + ); } if (persisted) { @@ -687,6 +699,27 @@ export const make = Effect.gen(function* () { } yield* installEntryLocked(entry, { retainEquivalentRuntime: true }); + + // The catalog entry carries no credential, so a fresh bearer keeps + // the equivalent runtime. Anything not yet connected may be using the + // old bearer: an attempt in flight would fail with it and then wait + // blocked for a manual retry. Retry with the new one instead. A + // connected session keeps its socket. + if (bearerReplaced) { + const scope = (yield* SubscriptionRef.get(serviceScopes)).get(target.environmentId); + if (scope !== undefined) { + const state = yield* SubscriptionRef.get(scope.supervisor.state); + if ( + state.phase === "connecting" || + state.phase === "backoff" || + (state.phase === "blocked" && + state.lastFailure?._tag === "ConnectionBlockedError" && + state.lastFailure.reason === "authentication") + ) { + yield* scope.supervisor.retryNow; + } + } + } }), ); }, -- 2.54.0