From 8fa10884a3eca07316c15e212635282104e961e2 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 29 Jul 2026 16:01:35 -0700 Subject: [PATCH] perf(mobile): speed up multi-environment reconnect on foreground Three serialization points made reconnecting to several environments feel sequential even though the supervisor fan-out is parallel: - The foreground liveness probe waited 15s on half-open sockets before starting a reconnect. It now times out after 5s. - Remote DPoP tokens lived inside the single connection-catalog SecureStore document, so every environment's token read/write serialized on one semaphore and rewrote the whole catalog to the Keychain. Tokens now live under per-environment keys with an in-memory cache and lazy migration. - The relay access-token cache held its lock across the network token exchange, so cache hits queued behind a stale account's round trip. Cache hits now take a lock-free fast path. Also reset the retry backoff ladder on application-active, since network conditions usually changed while backgrounded. Co-Authored-By: Claude Fable 5 --- apps/mobile/src/connection/catalog-store.ts | 2 +- apps/mobile/src/connection/storage.ts | 49 ++--- .../mobile/src/connection/token-store.test.ts | 177 ++++++++++++++++++ apps/mobile/src/connection/token-store.ts | 174 +++++++++++++++++ docs/architecture/connection-runtime.md | 2 + .../src/connection/supervisor.test.ts | 43 ++++- .../src/connection/supervisor.ts | 23 ++- .../src/relay/managedRelay.test.ts | 89 +++++++++ .../client-runtime/src/relay/managedRelay.ts | 33 +++- 9 files changed, 542 insertions(+), 50 deletions(-) create mode 100644 apps/mobile/src/connection/token-store.test.ts create mode 100644 apps/mobile/src/connection/token-store.ts diff --git a/apps/mobile/src/connection/catalog-store.ts b/apps/mobile/src/connection/catalog-store.ts index 6a4fcd35f6d..29bb1c9c1d1 100644 --- a/apps/mobile/src/connection/catalog-store.ts +++ b/apps/mobile/src/connection/catalog-store.ts @@ -41,7 +41,7 @@ const encodeCatalog = Effect.fn("mobile.connectionStorage.encodeCatalog")(functi ); }); -interface CatalogStore { +export interface CatalogStore { readonly read: Effect.Effect; readonly update: ( transform: (catalog: ConnectionCatalogDocumentType) => ConnectionCatalogDocumentType, diff --git a/apps/mobile/src/connection/storage.ts b/apps/mobile/src/connection/storage.ts index e844181673d..5af5d1d1613 100644 --- a/apps/mobile/src/connection/storage.ts +++ b/apps/mobile/src/connection/storage.ts @@ -9,7 +9,7 @@ import { } from "@t3tools/client-runtime/platform"; import { TokenStore } from "@t3tools/client-runtime/authorization"; import { - ConnectionTransientError, + type ConnectionAttemptError, CredentialStore, ProfileStore, } from "@t3tools/client-runtime/connection"; @@ -18,10 +18,11 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as CatalogStore from "./catalog-store"; +import * as RemoteDpopTokenStore from "./token-store"; function targetPersistenceError( operation: "list-targets" | "register-connection" | "remove-connection", - error: ConnectionTransientError, + error: ConnectionAttemptError, ) { return new ConnectionPersistenceError({ operation, @@ -32,6 +33,9 @@ function targetPersistenceError( export const connectionStorageLayer = Layer.effectContext( Effect.gen(function* () { const catalog = yield* CatalogStore.make(); + // Tokens live outside the catalog document so reconnecting environments + // do not serialize on the catalog lock. See token-store.ts. + const remoteTokenStore = yield* RemoteDpopTokenStore.make(catalog); const targetStore = ConnectionTargetStore.of({ list: catalog.read.pipe( @@ -44,10 +48,17 @@ export const connectionStorageLayer = Layer.effectContext( catalog .update((document) => registerConnectionInCatalog(document, registration)) .pipe(Effect.mapError((error) => targetPersistenceError("register-connection", error))), + // The token key is deleted before the catalog entry: if the keychain + // delete fails the registration survives and the removal can be retried, + // instead of leaving an orphaned token that a re-added environment + // would pick back up. remove: (target) => - catalog - .update((document) => removeConnectionFromCatalog(document, target)) - .pipe(Effect.mapError((error) => targetPersistenceError("remove-connection", error))), + remoteTokenStore.remove(target.environmentId).pipe( + Effect.andThen( + catalog.update((document) => removeConnectionFromCatalog(document, target)), + ), + Effect.mapError((error) => targetPersistenceError("remove-connection", error)), + ), }); const profileStore = ProfileStore.make({ get: (connectionId) => @@ -100,34 +111,6 @@ export const connectionStorageLayer = Layer.effectContext( ), })), }); - const remoteTokenStore = TokenStore.make({ - get: (environmentId) => - catalog.read.pipe( - Effect.map((document) => - Option.fromUndefinedOr( - document.remoteDpopTokens.find((token) => token.environmentId === environmentId), - ), - ), - ), - put: (token) => - catalog.update((document) => ({ - ...document, - remoteDpopTokens: replaceCatalogValue( - document.remoteDpopTokens, - (value) => value.environmentId, - token, - ), - })), - remove: (environmentId) => - catalog.update((document) => ({ - ...document, - remoteDpopTokens: removeCatalogValue( - document.remoteDpopTokens, - (value) => value.environmentId, - environmentId, - ), - })), - }); return Context.make(ConnectionTargetStore, targetStore).pipe( Context.add(ConnectionRegistrationStore, registrationStore), Context.add(ProfileStore.ConnectionProfileStore, profileStore), diff --git a/apps/mobile/src/connection/token-store.test.ts b/apps/mobile/src/connection/token-store.test.ts new file mode 100644 index 00000000000..45962fcbbac --- /dev/null +++ b/apps/mobile/src/connection/token-store.test.ts @@ -0,0 +1,177 @@ +import { TokenStore } from "@t3tools/client-runtime/authorization"; +import { EnvironmentId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Encoding from "effect/Encoding"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import { vi } from "vite-plus/test"; + +vi.mock("react-native", () => ({ + Platform: { OS: "ios" }, +})); + +vi.mock("expo-secure-store", () => ({ + deleteItemAsync: vi.fn(), + getItemAsync: vi.fn(), + setItemAsync: vi.fn(), +})); + +import * as CatalogStore from "./catalog-store"; +import { REMOTE_DPOP_TOKEN_KEY_PREFIX, make } from "./token-store"; +import { MobileSecureStorage } from "../persistence/mobile-secure-storage"; + +const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); + +const TOKEN = new TokenStore.RemoteDpopAccessToken({ + environmentId: ENVIRONMENT_ID, + label: "Remote", + endpoint: { + httpBaseUrl: "https://remote.example.test", + wsBaseUrl: "wss://remote.example.test", + providerKind: "cloudflare_tunnel", + }, + accessToken: "dpop-token", + expiresAtEpochMs: 1_000_000, + dpopThumbprint: "thumbprint", +}); + +const TOKEN_KEY = `${REMOTE_DPOP_TOKEN_KEY_PREFIX}${Encoding.encodeBase64Url(ENVIRONMENT_ID)}`; + +const encodeToken = Schema.encodeSync(TokenStore.RemoteDpopAccessToken); + +function makeStorage(initial: Readonly> = {}) { + const values = new Map(Object.entries(initial)); + const writes: Array = []; + const storage = MobileSecureStorage.of({ + getItem: (key) => Effect.sync(() => values.get(key) ?? null), + setItem: (key, value) => + Effect.sync(() => { + writes.push(key); + values.set(key, value); + }), + removeItem: (key) => + Effect.sync(() => { + values.delete(key); + }), + }); + return { storage, values, writes }; +} + +const makeStores = Effect.fn("makeStores")(function* (storage: MobileSecureStorage["Service"]) { + const catalog = yield* CatalogStore.make().pipe( + Effect.provideService(MobileSecureStorage, storage), + ); + const tokens = yield* make(catalog).pipe(Effect.provideService(MobileSecureStorage, storage)); + return { catalog, tokens }; +}); + +describe("mobile remote DPoP token store", () => { + it.effect("stores each token under its own key without touching the catalog", () => + Effect.gen(function* () { + const memory = makeStorage(); + const { catalog, tokens } = yield* makeStores(memory.storage); + + yield* tokens.put(TOKEN); + expect(memory.values.has(TOKEN_KEY)).toBe(true); + expect((yield* catalog.read).remoteDpopTokens).toEqual([]); + + const loaded = yield* tokens.get(ENVIRONMENT_ID); + expect(Option.getOrThrow(loaded)).toMatchObject({ accessToken: "dpop-token" }); + + yield* tokens.remove(ENVIRONMENT_ID); + expect(memory.values.has(TOKEN_KEY)).toBe(false); + expect(Option.isNone(yield* tokens.get(ENVIRONMENT_ID))).toBe(true); + }), + ); + + it.effect("serves repeat reads from memory without re-reading the keychain", () => + Effect.gen(function* () { + const memory = makeStorage(); + const { tokens } = yield* makeStores(memory.storage); + + yield* tokens.put(TOKEN); + // Corrupt the persisted copy: a cached read must not notice. + memory.values.set(TOKEN_KEY, "{not-json"); + const loaded = yield* tokens.get(ENVIRONMENT_ID); + expect(Option.getOrThrow(loaded).accessToken).toBe("dpop-token"); + }), + ); + + it.effect("migrates a token persisted inside the catalog document to its own key", () => + Effect.gen(function* () { + const memory = makeStorage(); + const { catalog, tokens } = yield* makeStores(memory.storage); + yield* catalog.update((document) => ({ + ...document, + remoteDpopTokens: [TOKEN], + })); + + const loaded = yield* tokens.get(ENVIRONMENT_ID); + expect(Option.getOrThrow(loaded)).toMatchObject({ accessToken: "dpop-token" }); + expect(memory.values.has(TOKEN_KEY)).toBe(true); + expect((yield* catalog.read).remoteDpopTokens).toEqual([]); + }), + ); + + it.effect("does not resurrect a token removed while a cold read is in flight", () => + Effect.gen(function* () { + const readGate = yield* Deferred.make(); + const readStarted = yield* Deferred.make(); + const memory = makeStorage(); + // getItem captures the stored value immediately (like a real Keychain + // read that already completed) but only delivers it after the gate, so + // the remove can run while the read is in flight. + const gatedStorage = MobileSecureStorage.of({ + getItem: (key) => + key === TOKEN_KEY + ? Effect.sync(() => memory.values.get(key) ?? null).pipe( + Effect.tap(() => Deferred.succeed(readStarted, undefined)), + Effect.tap(() => Deferred.await(readGate)), + ) + : Effect.sync(() => memory.values.get(key) ?? null), + setItem: (key, value) => + Effect.sync(() => { + memory.values.set(key, value); + }), + removeItem: (key) => + Effect.sync(() => { + memory.values.delete(key); + }), + }); + const { tokens } = yield* makeStores(gatedStorage); + yield* Effect.sync(() => memory.values.set(TOKEN_KEY, JSON.stringify(encodeToken(TOKEN)))); + + const pendingGet = yield* tokens.get(ENVIRONMENT_ID).pipe(Effect.forkScoped); + yield* Deferred.await(readStarted); + const pendingRemove = yield* tokens.remove(ENVIRONMENT_ID).pipe(Effect.forkScoped); + // Give remove every chance to run while the read is parked on the gate. + // With per-environment serialization it stays queued behind the get; a + // regression lets it delete the keychain entry here, and the resumed + // get would then resurrect the removed token into the cache. + for (let i = 0; i < 20 && memory.values.has(TOKEN_KEY); i += 1) { + yield* Effect.yieldNow; + } + yield* Deferred.succeed(readGate, undefined); + yield* Fiber.join(pendingGet); + yield* Fiber.join(pendingRemove); + + // remove ran after get completed (same-env operations serialize), so + // the token is gone from both the keychain and the in-memory cache. + expect(memory.values.has(TOKEN_KEY)).toBe(false); + expect(Option.isNone(yield* tokens.get(ENVIRONMENT_ID))).toBe(true); + }), + ); + + it.effect("discards a corrupt persisted token instead of failing the connection", () => + Effect.gen(function* () { + const memory = makeStorage({ [TOKEN_KEY]: "{not-json" }); + const { tokens } = yield* makeStores(memory.storage); + + expect(Option.isNone(yield* tokens.get(ENVIRONMENT_ID))).toBe(true); + expect(memory.values.has(TOKEN_KEY)).toBe(false); + }), + ); +}); diff --git a/apps/mobile/src/connection/token-store.ts b/apps/mobile/src/connection/token-store.ts new file mode 100644 index 00000000000..870e2e79e1e --- /dev/null +++ b/apps/mobile/src/connection/token-store.ts @@ -0,0 +1,174 @@ +import { TokenStore } from "@t3tools/client-runtime/authorization"; +import { ConnectionTransientError } from "@t3tools/client-runtime/connection"; +import type { EnvironmentId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Encoding from "effect/Encoding"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; + +import * as MobileSecureStorage from "../persistence/mobile-secure-storage"; +import type { CatalogStore } from "./catalog-store"; + +export const REMOTE_DPOP_TOKEN_KEY_PREFIX = "t3code.remote-dpop-token."; + +const RemoteDpopAccessTokenJson = Schema.fromJsonString(TokenStore.RemoteDpopAccessToken); +const decodeRemoteDpopAccessToken = Schema.decodeEffect(RemoteDpopAccessTokenJson); +const encodeRemoteDpopAccessToken = Schema.encodeEffect(RemoteDpopAccessTokenJson); + +function tokenError(operation: string, cause: unknown) { + return new ConnectionTransientError({ + reason: "remote-unavailable", + detail: `Could not ${operation} the environment access token: ${String(cause)}`, + }); +} + +// SecureStore only accepts [A-Za-z0-9._-] keys and the EnvironmentId schema +// allows free-form strings, so the id is base64url-encoded: collision-free, +// unlike lossy character replacement, which would let two ids share a key and +// silently clobber each other's tokens. +function tokenKey(environmentId: EnvironmentId): string { + return REMOTE_DPOP_TOKEN_KEY_PREFIX + Encoding.encodeBase64Url(environmentId); +} + +/** + * Remote DPoP access tokens used to live inside the connection catalog + * document, which put every token read and write behind one semaphore and + * re-encoded the whole catalog into a Keychain item on each update. With + * several environments reconnecting at once that serialized the parallel + * supervisor fan-out into O(N) full-document Keychain writes. + * + * Each token now lives under its own SecureStore key with a write-through + * in-memory cache, so per-environment token operations are independent. + * Tokens still found in the catalog document are migrated to their own key + * the first time the environment asks for them. + */ +export const make = Effect.fn("mobile.remoteDpopTokenStore.make")(function* ( + catalog: CatalogStore, +) { + const storage = yield* MobileSecureStorage.MobileSecureStorage; + const cache = yield* Ref.make(new Map()); + + // Operations on the same environment serialize so an in-flight `get` cannot + // republish a token that a concurrent `remove` or `put` just replaced. + // Different environments never share a lock; that independence is the point + // of this store. + const locks = new Map(); + const withEnvironmentLock = (environmentId: EnvironmentId) => { + let lock = locks.get(environmentId); + if (lock === undefined) { + lock = Semaphore.makeUnsafe(1); + locks.set(environmentId, lock); + } + return lock.withPermits(1); + }; + + const cachePut = (token: TokenStore.RemoteDpopAccessToken) => + Ref.update(cache, (entries) => new Map(entries).set(token.environmentId, token)); + const cacheRemove = (environmentId: EnvironmentId) => + Ref.update(cache, (entries) => { + const next = new Map(entries); + next.delete(environmentId); + return next; + }); + + const writeToken = Effect.fn("mobile.remoteDpopTokenStore.writeToken")(function* ( + token: TokenStore.RemoteDpopAccessToken, + ) { + const encoded = yield* encodeRemoteDpopAccessToken(token).pipe( + Effect.mapError((cause) => tokenError("encode", cause)), + ); + yield* storage + .setItem(tokenKey(token.environmentId), encoded) + .pipe(Effect.mapError((cause) => tokenError("save", cause))); + yield* cachePut(token); + }); + + // One-time move of a token stored in the catalog document (the pre-split + // layout) to its own key. Clearing the document entry costs one full-catalog + // write, but only for environments persisted before the split. + const migrateLegacyToken = Effect.fn("mobile.remoteDpopTokenStore.migrateLegacyToken")(function* ( + environmentId: EnvironmentId, + ) { + const document = yield* catalog.read; + const legacy = document.remoteDpopTokens.find((token) => token.environmentId === environmentId); + if (legacy === undefined) { + return Option.none(); + } + yield* writeToken(legacy); + // The standalone key is now authoritative, so a failed catalog cleanup + // must not fail the read; the leftover duplicate is purged on removal. + yield* catalog + .update((current) => ({ + ...current, + remoteDpopTokens: current.remoteDpopTokens.filter( + (token) => token.environmentId !== environmentId, + ), + })) + .pipe( + Effect.catch((error) => + Effect.logWarning("Could not clear a migrated environment access token", error), + ), + ); + return Option.some(legacy); + }); + + const getLocked = Effect.fn("mobile.remoteDpopTokenStore.get")(function* ( + environmentId: EnvironmentId, + ) { + const cached = (yield* Ref.get(cache)).get(environmentId); + if (cached !== undefined) { + return Option.some(cached); + } + const raw = yield* storage + .getItem(tokenKey(environmentId)) + .pipe(Effect.mapError((cause) => tokenError("load", cause))); + if (raw === null) { + return yield* migrateLegacyToken(environmentId); + } + const decoded = yield* decodeRemoteDpopAccessToken(raw).pipe( + Effect.mapError((cause) => tokenError("decode", cause)), + Effect.catch((error) => + Effect.logWarning("Discarding corrupt persisted environment access token", error).pipe( + Effect.andThen( + storage + .removeItem(tokenKey(environmentId)) + .pipe(Effect.mapError((cause) => tokenError("delete", cause))), + ), + Effect.as(null), + ), + ), + ); + if (decoded === null || decoded.environmentId !== environmentId) { + return Option.none(); + } + yield* cachePut(decoded); + return Option.some(decoded); + }); + + const removeLocked = Effect.fn("mobile.remoteDpopTokenStore.remove")(function* ( + environmentId: EnvironmentId, + ) { + yield* cacheRemove(environmentId); + yield* storage + .removeItem(tokenKey(environmentId)) + .pipe(Effect.mapError((cause) => tokenError("delete", cause))); + // Also drop any pre-split copy so it cannot resurface through migration. + const document = yield* catalog.read; + if (document.remoteDpopTokens.some((token) => token.environmentId === environmentId)) { + yield* catalog.update((current) => ({ + ...current, + remoteDpopTokens: current.remoteDpopTokens.filter( + (token) => token.environmentId !== environmentId, + ), + })); + } + }); + + return TokenStore.make({ + get: (environmentId) => withEnvironmentLock(environmentId)(getLocked(environmentId)), + put: (token) => withEnvironmentLock(token.environmentId)(writeToken(token)), + remove: (environmentId) => withEnvironmentLock(environmentId)(removeLocked(environmentId)), + }); +}); diff --git a/docs/architecture/connection-runtime.md b/docs/architecture/connection-runtime.md index 06f7e6338ca..dd8ffa9a622 100644 --- a/docs/architecture/connection-runtime.md +++ b/docs/architecture/connection-runtime.md @@ -40,6 +40,8 @@ The supervisor is the only retry owner. seconds. 5. Connectivity changes, application activation, credential changes, and explicit user retry interrupt the current wait and trigger a fresh attempt. + Application activation also resets the backoff ladder, since network + conditions usually changed while the app was backgrounded. 6. Authentication or configuration failures remain blocked until an external wakeup changes the relevant input. 7. An involuntary session close keeps the registration and cache, then retries. diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f3901e42251..b6cc0a0816d 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -652,6 +652,47 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("restarts the retry ladder when the application returns to the foreground", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* harness.closeLatestSession(); + yield* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + yield* TestClock.adjust("1 second"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + yield* harness.closeLatestSession(); + yield* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + + // Foregrounding interrupts the two-second backoff immediately and + // resets the ladder, so the next failure retries as attempt one again. + yield* harness.wake("application-active"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 3 && state.attempt === 1, + ); + yield* harness.closeLatestSession(); + yield* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + + expect(yield* Ref.get(harness.sessionCount)).toBe(3); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("probes the active session without reconnecting on application activation", () => Effect.gen(function* () { const probeCount = yield* Ref.make(0); @@ -712,7 +753,7 @@ describe("EnvironmentSupervisor", () => { yield* awaitState(supervisor.state, (state) => state.phase === "connected"); yield* harness.wake("application-active"); - yield* TestClock.adjust("15 seconds"); + yield* TestClock.adjust("5 seconds"); yield* awaitState( supervisor.state, (state) => state.phase === "backoff" && state.lastFailure?.reason === "timeout", diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d9efcd4263a..9a2d4d37894 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -31,7 +31,11 @@ import * as ConnectionWakeups from "./wakeups.ts"; const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const; const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; -const CONNECTION_PROBE_TIMEOUT = "15 seconds"; +// Deliberately short: the probe runs over an already-open socket, so a healthy +// server answers in well under a second. After mobile backgrounding the socket +// is often half-open (the OS dropped TCP without a close event) and every +// second spent waiting here is visible dead time before the reconnect starts. +const CONNECTION_PROBE_TIMEOUT = "5 seconds"; const BACKOFF_RESET_AFTER_MS = 30_000; interface SupervisorIntent { @@ -563,19 +567,24 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS); }, Effect.ensuring(clearLease)); + // Resolves when the backoff delay elapses or a signal interrupts the wait. + // Returns true when the app returned to the foreground: network conditions + // likely changed while backgrounded, so the caller restarts the retry ladder + // instead of resuming deep in it. const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) { return yield* Effect.raceFirst( - Effect.sleep(delayMs), + Effect.sleep(delayMs).pipe(Effect.as(false)), Effect.gen(function* () { for (;;) { const next = yield* Queue.take(signals); switch (next._tag) { + case "Wakeup": + return next.reason === "application-active"; case "ConnectRequested": case "DisconnectRequested": case "RetryRequested": case "NetworkChanged": - case "Wakeup": - return; + return false; } } }), @@ -663,7 +672,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( lastFailure: error, retryAt: (yield* Clock.currentTimeMillis) + delayMs, }); - yield* waitForRetrySignal(delayMs); + const foregrounded = yield* waitForRetrySignal(delayMs); + if (foregrounded) { + failureCount = 0; + pendingRetry = Option.none(); + } } }); diff --git a/packages/client-runtime/src/relay/managedRelay.test.ts b/packages/client-runtime/src/relay/managedRelay.test.ts index 278c205883f..b9aeaea348e 100644 --- a/packages/client-runtime/src/relay/managedRelay.test.ts +++ b/packages/client-runtime/src/relay/managedRelay.test.ts @@ -190,6 +190,95 @@ describe("ManagedRelayClient", () => { }).pipe(Effect.provide(managedRelayTestLayer(fetchFn))); }); + it.effect("serves cached tokens while another token exchange is still in flight", () => { + let releaseExchange = () => undefined as void; + const exchangeGate = new Promise((resolve) => { + releaseExchange = resolve; + }); + // Resolved from inside the exchange fetch handler, i.e. once the exchange + // fiber is parked inside the token cache critical section holding the lock. + let markExchangeStarted = () => undefined as void; + const exchangeStarted = new Promise((resolve) => { + markExchangeStarted = resolve; + }); + const persistedTokens: ReadonlyArray = [ + { + accountId: "user-cached", + clientId: "t3-mobile", + relayUrl: "https://relay.example.test", + thumbprint: "client-thumbprint", + scopes: [RelayEnvironmentStatusScope], + accessToken: "cached-relay-token", + expiresAtMillis: Number.MAX_SAFE_INTEGER, + }, + ]; + const accessTokenStore: ManagedRelay.ManagedRelayAccessTokenStore = { + load: Effect.succeed(persistedTokens), + save: () => Effect.void, + clear: Effect.void, + }; + const statusResponse = () => + Response.json({ + environmentId: "env-1", + endpoint: { + httpBaseUrl: "https://desktop.example.test/", + wsBaseUrl: "wss://desktop.example.test/ws", + providerKind: "cloudflare_tunnel", + }, + status: "online", + checkedAt: "2026-06-05T20:00:00.000Z", + descriptor: { + environmentId: "env-1", + label: "Desktop", + platform: { os: "darwin", arch: "arm64" }, + serverVersion: "0.0.0-test", + capabilities: { repositoryIdentity: true }, + }, + }); + const fetchFn = ((input) => { + const url = String(input); + if (url.endsWith("/v1/client/dpop-token")) { + // Stall the exchange until the cached-account request has finished. + markExchangeStarted(); + return exchangeGate.then(() => + Response.json({ + access_token: "fresh-relay-token", + issued_token_type: "urn:ietf:params:oauth:token-type:access_token", + token_type: "DPoP", + expires_in: 1_800, + scope: RelayEnvironmentStatusScope, + }), + ); + } + return Promise.resolve(statusResponse()); + }) satisfies typeof globalThis.fetch; + const statusInput = (token: string) => + ({ + clerkToken: token, + scopes: [RelayEnvironmentStatusScope], + environmentId: EnvironmentId.make("env-1"), + }) as const; + + return Effect.gen(function* () { + const relayClient = yield* ManagedRelay.ManagedRelayClient; + const pendingExchange = yield* relayClient + .getEnvironmentStatus(statusInput(clerkToken("user-fresh", "session-1"))) + .pipe(Effect.forkScoped); + // Wait until the exchange fiber holds the token cache lock; only then is + // the cached-account request proven not to queue behind it. Without the + // lock-free fast path this request deadlocks and the test times out. + yield* Effect.promise(() => exchangeStarted); + const cachedResult = yield* relayClient.getEnvironmentStatus( + statusInput(clerkToken("user-cached", "session-2")), + ); + expect(cachedResult.status).toBe("online"); + + releaseExchange(); + const freshResult = yield* Fiber.join(pendingExchange); + expect(freshResult.status).toBe("online"); + }).pipe(Effect.provide(managedRelayTestLayer(fetchFn, undefined, accessTokenStore))); + }); + it.effect("reuses a persisted token across runtimes and Clerk session token rotation", () => { let tokenExchangeCount = 0; let persistedTokens: ReadonlyArray = []; diff --git a/packages/client-runtime/src/relay/managedRelay.ts b/packages/client-runtime/src/relay/managedRelay.ts index 06769a4372f..f188e463bb4 100644 --- a/packages/client-runtime/src/relay/managedRelay.ts +++ b/packages/client-runtime/src/relay/managedRelay.ts @@ -546,19 +546,32 @@ export const make = Effect.fn("ManagedRelayClient.make")(function* ( expiresAtMillis: nowMillis + response.expires_in * 1_000, } satisfies ManagedRelayAccessTokenCacheEntry; } + const matchInput = { + accountId: accountId.value, + clientId: options.clientId, + relayUrl, + thumbprint: input.thumbprint, + scopes: input.scopes, + nowMillis, + }; + // Fast path outside the critical section: many environments authorize + // concurrently on resume, and a cache hit must not wait behind another + // fiber's in-flight token exchange. + const fastPath = (yield* SynchronizedRef.get(cachedTokens)).find((token) => + tokenMatches(token, matchInput), + ); + if (fastPath) { + yield* Effect.annotateCurrentSpan({ + "relay.token_cache.result": "hit", + }); + return fastPath; + } + // The critical section re-checks the cache so concurrent misses share + // one exchange: the first fiber performs it, the rest find its result. return yield* SynchronizedRef.modifyEffect(cachedTokens, (tokens) => Effect.gen(function* () { const activeTokens = tokens.filter((token) => token.expiresAtMillis > nowMillis + 5_000); - const cached = activeTokens.find((token) => - tokenMatches(token, { - accountId: accountId.value, - clientId: options.clientId, - relayUrl, - thumbprint: input.thumbprint, - scopes: input.scopes, - nowMillis, - }), - ); + const cached = activeTokens.find((token) => tokenMatches(token, matchInput)); if (cached) { yield* Effect.annotateCurrentSpan({ "relay.token_cache.result": "hit",