From 703c7c295f2e2aabd38e323d178bca9aeb4775ec Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:04:25 -0800 Subject: [PATCH] fix(client-runtime): a stalled sign-in token or dead route no longer holds a reconnect for 40 seconds A Clerk token read that has not answered after 4 seconds is replaced with a fresh one, up to three tries. A route that stayed silent for its whole reachability check is skipped once another route failed in a way that retries. Co-Authored-By: Claude Fable 5.2 --- .../client-runtime/src/connection/driver.ts | 14 +++--- .../src/connection/supervisor.test.ts | 30 +++++++++++++ .../src/relay/managedRelayState.test.ts | 37 ++++++++++++++++ .../src/relay/managedRelayState.ts | 44 +++++++++++++------ 4 files changed, 105 insertions(+), 20 deletions(-) diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index 45f12d0b74509..6271f4e4e5caa 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -115,16 +115,16 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") // An incompatible server is the same server on every route. if (result.failure.reason === "unsupported") return yield* result.failure; } - // 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"), - ]; - for (const { route } of retry) { + const answeredLate = silent.filter((_, position) => late[position] !== "silent"); + const stillSilent = silent.filter((_, position) => late[position] === "silent"); + for (const { route } of [...answeredLate, ...stillSilent]) { + // A route still silent after its whole check can hold the attempt open far + // 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); 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 f284c48e82772..a502db52697fc 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -1640,6 +1640,36 @@ describe("EnvironmentSupervisor routes", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("retries instead of waiting on a LAN route that stayed silent after relay failed", () => + Effect.gen(function* () { + const checkingLan = yield* Deferred.make(); + const attempted: string[] = []; + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Deferred.succeed(checkingLan, undefined).pipe( + Effect.andThen(Effect.sleep("2500 millis")), + Effect.as("silent" as const), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => { + attempted.push(target._tag); + return target._tag === "RelayConnectionTarget" + ? Effect.fail(transient("Relay socket closed.")) + : Effect.never; + }, + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(checkingLan); + yield* TestClock.adjust("2500 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "backoff"); + expect(attempted).toEqual(["RelayConnectionTarget"]); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("skips a silent LAN route and connects over T3 Connect", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/relay/managedRelayState.test.ts b/packages/client-runtime/src/relay/managedRelayState.test.ts index 6b91a2213bbe7..41d7327ac14e4 100644 --- a/packages/client-runtime/src/relay/managedRelayState.test.ts +++ b/packages/client-runtime/src/relay/managedRelayState.test.ts @@ -8,6 +8,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; +import * as TestClock from "effect/testing/TestClock"; import { Atom, AtomRegistry } from "effect/reactivity"; import { afterEach, vi } from "vite-plus/test"; @@ -191,6 +192,42 @@ describe("createManagedRelayQueryManager", () => { }), ); + it.effect("replaces a Clerk token read that has not answered after 4 seconds", () => + Effect.gen(function* () { + const token = clerkToken(4_102_444_800); + const readClerkToken = vi + .fn<() => Promise>() + .mockReturnValueOnce(new Promise(() => {})) + .mockResolvedValue(token); + const session = createManagedRelaySession({ accountId: "account-1", readClerkToken }); + + const readsFiber = yield* Effect.all([session.readClerkToken(), session.readClerkToken()], { + concurrency: "unbounded", + }).pipe(Effect.forkChild); + yield* TestClock.adjust("3999 millis"); + expect(readClerkToken).toHaveBeenCalledTimes(1); + + yield* TestClock.adjust("1 millis"); + expect(yield* Fiber.join(readsFiber)).toEqual([token, token]); + expect(readClerkToken).toHaveBeenCalledTimes(2); + }), + ); + + it.effect("fails a Clerk token read that never answers after three tries", () => + Effect.gen(function* () { + const readClerkToken = vi.fn(() => new Promise(() => {})); + const session = createManagedRelaySession({ accountId: "account-1", readClerkToken }); + + const readFiber = yield* session.readClerkToken().pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("12 seconds"); + + expect((yield* Fiber.join(readFiber)).message).toBe( + "Timed out obtaining the T3 Connect session token.", + ); + expect(readClerkToken).toHaveBeenCalledTimes(3); + }), + ); + it.effect("updates the token provider without replacing a same-account session", () => Effect.gen(function* () { const firstRead = vi.fn(() => Promise.resolve(null)); diff --git a/packages/client-runtime/src/relay/managedRelayState.ts b/packages/client-runtime/src/relay/managedRelayState.ts index 5a4b9a62a547a..90e635f4e328a 100644 --- a/packages/client-runtime/src/relay/managedRelayState.ts +++ b/packages/client-runtime/src/relay/managedRelayState.ts @@ -3,6 +3,7 @@ import { decodeRelayJwt } from "@t3tools/shared/relayJwt"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; @@ -14,6 +15,11 @@ import { relayProtectedErrorMessage } from "./errorPresentation.ts"; const DEFAULT_STALE_TIME_MS = 15_000; const DEFAULT_IDLE_TTL_MS = 5 * 60_000; const CLERK_TOKEN_EXPIRY_SKEW_MS = 5_000; +// Clerk retries a failed token request itself, but waits 3 s before the first +// retry and longer before each next one. A fresh request usually answers at +// once, so a slow read is replaced instead of waited out. +const CLERK_TOKEN_READ_TIMEOUT_MS = 4_000; +const CLERK_TOKEN_READ_ATTEMPTS = 3; const isManagedRelayRequestFailedError = Schema.is(ManagedRelay.ManagedRelayRequestFailedError); export interface ManagedRelaySession { @@ -66,7 +72,10 @@ const managedRelaySessionControls = new WeakMap | null = null; + let pendingToken: { + readonly operation: Promise; + readonly startedAtMillis: number; + } | null = null; let readClerkToken = input.readClerkToken; let tokenProviderGeneration = 0; @@ -74,8 +83,10 @@ export function createManagedRelaySession(input: ManagedRelaySessionInput): Mana if (cachedToken && cachedToken.expiresAtMillis > nowMillis + CLERK_TOKEN_EXPIRY_SKEW_MS) { return cachedToken.token; } - if (pendingToken) { - return await pendingToken; + // A read that outlived its timeout is left running, since it still fills + // the cache, but later readers start a fresh one instead of joining it. + if (pendingToken && nowMillis - pendingToken.startedAtMillis < CLERK_TOKEN_READ_TIMEOUT_MS) { + return await pendingToken.operation; } const operationGeneration = tokenProviderGeneration; @@ -99,11 +110,12 @@ export function createManagedRelaySession(input: ManagedRelaySessionInput): Mana : null; return token; }); - pendingToken = operation; + const pending = { operation, startedAtMillis: nowMillis }; + pendingToken = pending; try { return await operation; } finally { - if (pendingToken === operation) { + if (pendingToken === pending) { pendingToken = null; } } @@ -112,14 +124,20 @@ export function createManagedRelaySession(input: ManagedRelaySessionInput): Mana const session: ManagedRelaySession = { accountId: input.accountId, readClerkToken: Effect.fn("clientRuntime.managedRelaySession.readClerkToken")(function* () { - const nowMillis = yield* Clock.currentTimeMillis; - return yield* Effect.tryPromise({ - try: () => readCachedClerkToken(nowMillis), - catch: (cause) => - new ManagedRelaySessionError({ - message: "Could not obtain the T3 Connect session token.", - cause, - }), + for (let attempt = 0; attempt < CLERK_TOKEN_READ_ATTEMPTS; attempt += 1) { + const nowMillis = yield* Clock.currentTimeMillis; + const token = yield* Effect.tryPromise({ + try: () => readCachedClerkToken(nowMillis), + catch: (cause) => + new ManagedRelaySessionError({ + message: "Could not obtain the T3 Connect session token.", + cause, + }), + }).pipe(Effect.timeoutOption(CLERK_TOKEN_READ_TIMEOUT_MS)); + if (Option.isSome(token)) return token.value; + } + return yield* new ManagedRelaySessionError({ + message: "Timed out obtaining the T3 Connect session token.", }); }), };