From 43b718a7446c8e07fb6a0f05f36587b14764fdb4 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 29 Jul 2026 14:59:50 -0700 Subject: [PATCH 1/6] perf(mobile): reconnect environments immediately on resume --- .../src/connection/app-state-wakeups.test.ts | 21 ++++++ .../src/connection/app-state-wakeups.ts | 18 +++++ apps/mobile/src/connection/platform.ts | 49 ++++++++++--- .../src/authorization/layer.test.ts | 49 +++++++++---- .../src/authorization/service.ts | 68 +++++++++---------- .../src/connection/supervisor.test.ts | 36 ++++++++-- .../src/connection/supervisor.ts | 17 ++++- .../client-runtime/src/connection/wakeups.ts | 6 +- .../src/state/shell-sync.test.ts | 8 +++ .../src/state/threads-sync.test.ts | 7 ++ 10 files changed, 211 insertions(+), 68 deletions(-) create mode 100644 apps/mobile/src/connection/app-state-wakeups.test.ts create mode 100644 apps/mobile/src/connection/app-state-wakeups.ts diff --git a/apps/mobile/src/connection/app-state-wakeups.test.ts b/apps/mobile/src/connection/app-state-wakeups.test.ts new file mode 100644 index 00000000000..4e8bdf1edf4 --- /dev/null +++ b/apps/mobile/src/connection/app-state-wakeups.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + MOBILE_BACKGROUND_RECONNECT_AFTER_MS, + mobileApplicationActiveWakeup, +} from "./app-state-wakeups"; + +describe("mobileApplicationActiveWakeup", () => { + it("uses a fast probe after a short interruption", () => { + expect(mobileApplicationActiveWakeup(null, 20_000)).toBe("application-active-probe"); + expect( + mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS - 1), + ).toBe("application-active-probe"); + }); + + it("replaces the session after a meaningful background suspension", () => { + expect( + mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS), + ).toBe("application-active-reconnect"); + }); +}); diff --git a/apps/mobile/src/connection/app-state-wakeups.ts b/apps/mobile/src/connection/app-state-wakeups.ts new file mode 100644 index 00000000000..185ff744958 --- /dev/null +++ b/apps/mobile/src/connection/app-state-wakeups.ts @@ -0,0 +1,18 @@ +import type { Wakeups } from "@t3tools/client-runtime/connection"; + +export const MOBILE_BACKGROUND_RECONNECT_AFTER_MS = 10_000; + +export type MobileApplicationActiveWakeup = Extract< + Wakeups.ConnectionWakeup, + "application-active-probe" | "application-active-reconnect" +>; + +export function mobileApplicationActiveWakeup( + backgroundedAtMs: number | null, + activeAtMs: number, +): MobileApplicationActiveWakeup { + return backgroundedAtMs !== null && + activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS + ? "application-active-reconnect" + : "application-active-probe"; +} diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index b8a13137c88..852535d9d10 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -30,6 +30,7 @@ import * as MobileStorage from "../persistence/mobile-storage"; import { appAtomRegistry } from "../state/atom-registry"; import { clearThreadOutboxEnvironment } from "../state/thread-outbox"; import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts"; +import { mobileApplicationActiveWakeup } from "./app-state-wakeups"; import { connectionStorageLayer } from "./storage"; function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" { @@ -54,27 +55,53 @@ const connectivityLayer = Connectivity.layer({ ), changes: Stream.callback((queue) => Effect.acquireRelease( - Effect.sync(() => - Network.addNetworkStateListener((state) => { + Effect.sync(() => { + let active = true; + const networkSubscription = Network.addNetworkStateListener((state) => { Queue.offerUnsafe(queue, networkStatus(state)); - }), - ), - (subscription) => Effect.sync(() => subscription.remove()), + }); + const appStateSubscription = AppState.addEventListener("change", (state) => { + if (state !== "active") { + return; + } + void Network.getNetworkStateAsync() + .then((current) => { + if (active) { + Queue.offerUnsafe(queue, networkStatus(current)); + } + }) + .catch(() => undefined); + }); + return { + close: () => { + active = false; + networkSubscription.remove(); + appStateSubscription.remove(); + }, + }; + }), + ({ close }) => Effect.sync(close), ).pipe(Effect.asVoid), ), }); const wakeupsLayer = Wakeups.layer({ changes: Stream.merge( - Stream.callback<"application-active">((queue) => + Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) => Effect.acquireRelease( - Effect.sync(() => - AppState.addEventListener("change", (state) => { + Effect.sync(() => { + let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null; + return AppState.addEventListener("change", (state) => { + if (state === "background") { + backgroundedAtMs = Date.now(); + return; + } if (state === "active") { - Queue.offerUnsafe(queue, "application-active"); + Queue.offerUnsafe(queue, mobileApplicationActiveWakeup(backgroundedAtMs, Date.now())); + backgroundedAtMs = null; } - }), - ), + }); + }), (subscription) => Effect.sync(() => subscription.remove()), ).pipe(Effect.asVoid), ), diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 1d2c6c6cca7..e47ad3a87b2 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -155,6 +155,39 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( }); describe("RemoteEnvironmentAuthorization", () => { + it.effect("reuses a validated bearer descriptor while issuing fresh websocket tickets", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + responses: [ + Response.json(DESCRIPTOR), + websocketTicket("first-ticket"), + websocketTicket("second-ticket"), + ], + }); + + const [first, second] = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + const authorize = () => + remote.authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + bearerToken: "bearer-token", + }); + return [yield* authorize(), yield* authorize()] as const; + }).pipe(Effect.provide(harness.layer)); + + expect(first.socketUrl).toContain("wsTicket=first-ticket"); + expect(second.socketUrl).toContain("wsTicket=second-ticket"); + expect( + harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")), + ).toHaveLength(1); + expect( + harness.fetch.calls.filter(([url]) => String(url).endsWith("/api/auth/websocket-ticket")), + ).toHaveLength(2); + }), + ); + it.effect("reuses a valid persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ @@ -265,7 +298,7 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); - it.effect("refreshes a cached endpoint after consecutive transient failures", () => + it.effect("refreshes a cached endpoint after its first transient failure", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ environmentId: ENVIRONMENT_ID, @@ -279,7 +312,6 @@ describe("RemoteEnvironmentAuthorization", () => { initialToken: cached, responses: [ new Response("endpoint unavailable", { status: 503 }), - new Response("endpoint still unavailable", { status: 503 }), Response.json(DESCRIPTOR), accessToken("replacement-access-token"), websocketTicket("replacement-ticket"), @@ -288,17 +320,6 @@ describe("RemoteEnvironmentAuthorization", () => { const authorized = yield* Effect.gen(function* () { const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; - const firstFailure = yield* remote - .authorizeDpop({ - expectedEnvironmentId: ENVIRONMENT_ID, - obtainBootstrap: harness.obtainBootstrap, - }) - .pipe(Effect.flip); - - expect(firstFailure._tag).toBe("ConnectionTransientError"); - expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); - expect((yield* Ref.get(harness.tokens)).get(ENVIRONMENT_ID)).toBe(cached); - return yield* remote.authorizeDpop({ expectedEnvironmentId: ENVIRONMENT_ID, obtainBootstrap: harness.obtainBootstrap, @@ -312,7 +333,7 @@ describe("RemoteEnvironmentAuthorization", () => { accessToken: "replacement-access-token", }), ); - expect(harness.fetch.calls).toHaveLength(5); + expect(harness.fetch.calls).toHaveLength(4); }), ); diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 624ecf7f672..ec517c98f56 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -1,4 +1,4 @@ -import { EnvironmentId } from "@t3tools/contracts"; +import { EnvironmentId, type ExecutionEnvironmentDescriptor } from "@t3tools/contracts"; import type { RelayManagedEndpoint } from "@t3tools/contracts/relay"; import { exchangeRemoteDpopAccessToken, @@ -58,7 +58,7 @@ export class RemoteEnvironmentAuthorization extends Context.Service< >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {} const TOKEN_EXPIRY_SAFETY_MARGIN_MS = 60_000; -const CACHED_ENDPOINT_FAILURE_THRESHOLD = 2; +const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) { return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError" @@ -79,25 +79,15 @@ export const make = Effect.gen(function* () { const presentation = yield* ClientCapabilities.ClientPresentation; const tokenStore = yield* TokenStore.RemoteDpopAccessTokenStore; const httpClient = yield* HttpClient.HttpClient; - const cachedEndpointFailures = yield* Ref.make>(new Map()); - - const resetCachedEndpointFailures = (environmentId: string) => - Ref.update(cachedEndpointFailures, (current) => { - if (!current.has(environmentId)) { - return current; + const bearerDescriptors = yield* Ref.make< + ReadonlyMap< + EnvironmentId, + { + readonly httpBaseUrl: string; + readonly descriptor: ExecutionEnvironmentDescriptor; } - const next = new Map(current); - next.delete(environmentId); - return next; - }); - - const recordCachedEndpointFailure = (environmentId: string) => - Ref.modify(cachedEndpointFailures, (current) => { - const failureCount = (current.get(environmentId) ?? 0) + 1; - const next = new Map(current); - next.set(environmentId, failureCount); - return [failureCount, next] as const; - }); + > + >(new Map()); const authorizeBearer = Effect.fn("clientRuntime.connection.remote.authorizeBearer")( function* (input: { @@ -108,15 +98,29 @@ export const make = Effect.gen(function* () { readonly wsBaseUrl: string; readonly bearerToken: string; }) { - const descriptor = yield* fetchDescriptor(input.httpBaseUrl).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const cachedDescriptor = (yield* Ref.get(bearerDescriptors)).get(input.expectedEnvironmentId); + const descriptor = + cachedDescriptor?.httpBaseUrl === input.httpBaseUrl + ? cachedDescriptor.descriptor + : yield* fetchDescriptor(input.httpBaseUrl).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, actual: descriptor.environmentId, }); } + if (cachedDescriptor?.httpBaseUrl !== input.httpBaseUrl) { + yield* Ref.update(bearerDescriptors, (current) => { + const next = new Map(current); + next.set(input.expectedEnvironmentId, { + httpBaseUrl: input.httpBaseUrl, + descriptor, + }); + return next; + }); + } const socketUrl = yield* resolveRemoteWebSocketConnectionUrl({ wsBaseUrl: input.wsBaseUrl, httpBaseUrl: input.httpBaseUrl, @@ -139,7 +143,7 @@ export const make = Effect.gen(function* () { ); const createDpopSocketUrl = Effect.fn("clientRuntime.connection.remote.createDpopSocketUrl")( - function* (token: TokenStore.RemoteDpopAccessToken) { + function* (token: TokenStore.RemoteDpopAccessToken, timeoutMs?: number) { const ticketProof = yield* signer .createProof({ method: "POST", @@ -160,6 +164,7 @@ export const make = Effect.gen(function* () { httpBaseUrl: token.endpoint.httpBaseUrl, accessToken: token.accessToken, dpopProof: ticketProof, + ...(timeoutMs === undefined ? {} : { timeoutMs }), }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient)); }, ); @@ -196,9 +201,11 @@ export const make = Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ "connection.remote_token_cache": "hit", }); - const cachedSocket = yield* createDpopSocketUrl(cached.value).pipe(Effect.result); + const cachedSocket = yield* createDpopSocketUrl( + cached.value, + CACHED_ENDPOINT_SOCKET_TIMEOUT_MS, + ).pipe(Effect.result); if (Result.isSuccess(cachedSocket)) { - yield* resetCachedEndpointFailures(input.expectedEnvironmentId); return { environmentId: cached.value.environmentId, label: cached.value.label, @@ -213,20 +220,11 @@ export const make = Effect.gen(function* () { if (cachedSocket.failure._tag === "ConnectionBlockedError") { return yield* mapDpopSocketError(cachedSocket.failure); } - const mappedFailure = mapDpopSocketError(cachedSocket.failure); - if (mappedFailure._tag === "ConnectionTransientError") { - const failureCount = yield* recordCachedEndpointFailure(input.expectedEnvironmentId); - if (failureCount < CACHED_ENDPOINT_FAILURE_THRESHOLD) { - return yield* mappedFailure; - } - } yield* tokenStore .remove(input.expectedEnvironmentId) .pipe(Effect.withSpan("environment.authorization.accessToken.remove")); - yield* resetCachedEndpointFailures(input.expectedEnvironmentId); } - yield* resetCachedEndpointFailures(input.expectedEnvironmentId); yield* Effect.annotateCurrentSpan({ "connection.remote_token_cache": "miss", }); diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f3901e42251..f935523caf4 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -124,7 +124,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: const releaseCount = yield* Ref.make(0); const wakeups = yield* SubscriptionRef.make<{ readonly sequence: number; - readonly reason: "application-active" | "credentials-changed"; + readonly reason: ConnectionWakeups.ConnectionWakeup; }>({ sequence: 0, reason: "application-active", @@ -198,7 +198,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: sessionCount, releaseCount, setNetworkStatus: (status: NetworkStatus) => SubscriptionRef.set(networkStatus, status), - wake: (reason: "application-active" | "credentials-changed") => + wake: (reason: ConnectionWakeups.ConnectionWakeup) => SubscriptionRef.update(wakeups, (event) => ({ sequence: event.sequence + 1, reason, @@ -677,6 +677,32 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("immediately replaces a mobile session after a long background resume", () => + Effect.gen(function* () { + const probeCount = yield* Ref.make(0); + const harness = yield* makeHarness({ + probe: () => Ref.update(probeCount, (count) => count + 1), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + + expect(yield* Ref.get(probeCount)).toBe(0); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }), + ); + it.effect("reconnects when the foreground liveness probe fails", () => Effect.gen(function* () { const harness = yield* makeHarness({ @@ -701,7 +727,7 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); - it.effect("times out a stalled foreground liveness probe and reconnects", () => + it.effect("quickly times out a stalled mobile foreground liveness probe", () => Effect.gen(function* () { const harness = yield* makeHarness({ probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), @@ -711,8 +737,8 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(harness.dependencies)); yield* awaitState(supervisor.state, (state) => state.phase === "connected"); - yield* harness.wake("application-active"); - yield* TestClock.adjust("15 seconds"); + yield* harness.wake("application-active-probe"); + yield* TestClock.adjust("3 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..80ea3587191 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -32,6 +32,7 @@ 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"; +const MOBILE_CONNECTION_PROBE_TIMEOUT = "3 seconds"; const BACKOFF_RESET_AFTER_MS = 30_000; interface SupervisorIntent { @@ -374,6 +375,9 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( case "ConnectRequested": break; case "Wakeup": + if (next.reason === "application-active-reconnect") { + return; + } if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") { yield* logManagedRelayAccountChange; return; @@ -402,10 +406,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( yield* logManagedRelayAccountChange; return; } - if (next.reason === "application-active") { + if (next.reason === "application-active-reconnect") { + // Mobile operating systems commonly suspend sockets without + // delivering a close event. A long background resume deliberately + // replaces that lease and starts a fresh attempt without backoff. + return; + } + if (next.reason === "application-active" || next.reason === "application-active-probe") { const probe = yield* lease.session.probe.pipe( Effect.timeoutOrElse({ - duration: CONNECTION_PROBE_TIMEOUT, + duration: + next.reason === "application-active-probe" + ? MOBILE_CONNECTION_PROBE_TIMEOUT + : CONNECTION_PROBE_TIMEOUT, orElse: () => Effect.fail( new ConnectionTransientError({ diff --git a/packages/client-runtime/src/connection/wakeups.ts b/packages/client-runtime/src/connection/wakeups.ts index 107c5983e02..cac96772100 100644 --- a/packages/client-runtime/src/connection/wakeups.ts +++ b/packages/client-runtime/src/connection/wakeups.ts @@ -2,7 +2,11 @@ import * as Context from "effect/Context"; import * as Layer from "effect/Layer"; import type * as Stream from "effect/Stream"; -export type ConnectionWakeup = "application-active" | "credentials-changed"; +export type ConnectionWakeup = + | "application-active" + | "application-active-probe" + | "application-active-reconnect" + | "credentials-changed"; export class ConnectionWakeups extends Context.Service< ConnectionWakeups, diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 62d2a2c28b9..d00298958cc 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -329,6 +329,14 @@ describe("environment shell synchronization", () => { expect(yield* Ref.get(loaderCalls)).toBe(2); expect(yield* Ref.get(subscriptionCount)).toBe(2); + + yield* Queue.offer(wakeups, "application-active-probe"); + yield* Queue.offer(wakeups, "application-active-reconnect"); + for (let attempt = 0; attempt < 10; attempt += 1) { + yield* Effect.yieldNow; + } + expect(yield* Ref.get(loaderCalls)).toBe(2); + expect(yield* Ref.get(subscriptionCount)).toBe(2); }), ); }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 43cdc1590bf..0b14941842c 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -696,6 +696,13 @@ describe("EnvironmentThreads", () => { (value) => value.status === "live" && Option.isSome(value.data), ); expect(Option.getOrThrow(live.data).title).toBe("Latest title"); + + yield* Queue.offer(harness.wakeups, "application-active-probe"); + yield* Queue.offer(harness.wakeups, "application-active-reconnect"); + for (let attempt = 0; attempt < 10; attempt += 1) { + yield* Effect.yieldNow; + } + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); }), ); }); From 29b67a087083458fbb61704010ed465d8fd567af Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 29 Jul 2026 15:03:21 -0700 Subject: [PATCH 2/6] fix(mobile): honor reconnect during active probe --- .../src/connection/supervisor.test.ts | 26 +++++++++++++++++++ .../src/connection/supervisor.ts | 11 +++++++- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f935523caf4..c4196690bb4 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -703,6 +703,32 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("replaces a mobile session when a long resume interrupts an active probe", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + yield* harness.wake("application-active-probe"); + yield* Effect.yieldNow; + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }), + ); + it.effect("reconnects when the foreground liveness probe fails", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 80ea3587191..1359e2359cc 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -453,8 +453,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( return; } break; - case "ConnectRequested": case "Wakeup": + if ( + probeEvent.signal.reason === "application-active-reconnect" || + (probeEvent.signal.reason === "credentials-changed" && + target._tag === "RelayConnectionTarget") + ) { + yield* Fiber.interrupt(probe); + return; + } + break; + case "ConnectRequested": break; } } From 86351226350a09c3d427daa36a384b16798c720c Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 30 Jul 2026 02:18:12 -0700 Subject: [PATCH 3/6] perf(mobile): reset reconnect backoff on foreground --- docs/architecture/connection-runtime.md | 3 ++ .../src/connection/supervisor.test.ts | 39 +++++++++++++++++++ .../src/connection/supervisor.ts | 13 +++++-- .../client-runtime/src/connection/wakeups.ts | 12 ++++++ .../src/state/shell-sync.test.ts | 11 +++++- packages/client-runtime/src/state/shell.ts | 2 +- .../src/state/threads-sync.test.ts | 8 +++- packages/client-runtime/src/state/threads.ts | 2 +- 8 files changed, 81 insertions(+), 9 deletions(-) diff --git a/docs/architecture/connection-runtime.md b/docs/architecture/connection-runtime.md index 06f7e6338ca..acfd2ab5983 100644 --- a/docs/architecture/connection-runtime.md +++ b/docs/architecture/connection-runtime.md @@ -40,6 +40,9 @@ 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. Mobile briefly probes + a session after short interruptions and replaces it immediately after a + meaningful background suspension. 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 c4196690bb4..5d783996b3b 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -652,6 +652,45 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("restarts the retry ladder when mobile 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, + ); + + yield* harness.wake("application-active-reconnect"); + 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); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 1359e2359cc..61b824d2b1c 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -587,17 +587,18 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( 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 ConnectionWakeups.isApplicationActiveWakeup(next.reason); case "ConnectRequested": case "DisconnectRequested": case "RetryRequested": case "NetworkChanged": - case "Wakeup": - return; + return false; } } }), @@ -685,7 +686,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( lastFailure: error, retryAt: (yield* Clock.currentTimeMillis) + delayMs, }); - yield* waitForRetrySignal(delayMs); + const applicationActivated = yield* waitForRetrySignal(delayMs); + if (applicationActivated) { + failureCount = 0; + pendingRetry = Option.none(); + } } }); diff --git a/packages/client-runtime/src/connection/wakeups.ts b/packages/client-runtime/src/connection/wakeups.ts index cac96772100..8573a49c147 100644 --- a/packages/client-runtime/src/connection/wakeups.ts +++ b/packages/client-runtime/src/connection/wakeups.ts @@ -8,6 +8,18 @@ export type ConnectionWakeup = | "application-active-reconnect" | "credentials-changed"; +export function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean { + return ( + reason === "application-active" || + reason === "application-active-probe" || + reason === "application-active-reconnect" + ); +} + +export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean { + return reason === "application-active" || reason === "application-active-probe"; +} + export class ConnectionWakeups extends Context.Service< ConnectionWakeups, { diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index d00298958cc..e006fc3cd76 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -331,12 +331,19 @@ describe("environment shell synchronization", () => { expect(yield* Ref.get(subscriptionCount)).toBe(2); yield* Queue.offer(wakeups, "application-active-probe"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(subscriptionCount)) >= 3) break; + yield* Effect.yieldNow; + } + expect(yield* Ref.get(loaderCalls)).toBe(3); + expect(yield* Ref.get(subscriptionCount)).toBe(3); + yield* Queue.offer(wakeups, "application-active-reconnect"); for (let attempt = 0; attempt < 10; attempt += 1) { yield* Effect.yieldNow; } - expect(yield* Ref.get(loaderCalls)).toBe(2); - expect(yield* Ref.get(subscriptionCount)).toBe(2); + expect(yield* Ref.get(loaderCalls)).toBe(3); + expect(yield* Ref.get(subscriptionCount)).toBe(3); }), ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 6ccb11797f5..a266af5f5f4 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -172,7 +172,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const foregroundResubscriptions = Option.match(wakeups, { onNone: () => Stream.never, onSome: (service) => - service.changes.pipe(Stream.filter((reason) => reason === "application-active")), + service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)), }); yield* setSynchronizing; diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 0b14941842c..c2df434e8e7 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -698,11 +698,17 @@ describe("EnvironmentThreads", () => { expect(Option.getOrThrow(live.data).title).toBe("Latest title"); yield* Queue.offer(harness.wakeups, "application-active-probe"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 3) break; + yield* Effect.yieldNow; + } + expect(yield* Ref.get(harness.subscriptionCount)).toBe(3); + yield* Queue.offer(harness.wakeups, "application-active-reconnect"); for (let attempt = 0; attempt < 10; attempt += 1) { yield* Effect.yieldNow; } - expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(3); }), ); }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 196229cc8b1..06b5428ca58 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -236,7 +236,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const foregroundResubscriptions = Option.match(wakeups, { onNone: () => Stream.never, onSome: (service) => - service.changes.pipe(Stream.filter((reason) => reason === "application-active")), + service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)), }); yield* setSynchronizing; From ef970927540083720af30212038c3d0b21ac8cc9 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 30 Jul 2026 02:25:15 -0700 Subject: [PATCH 4/6] fix(client-runtime): expire cached environment descriptors --- .../src/authorization/layer.test.ts | 43 +++++++++++++++++++ .../src/authorization/service.ts | 20 ++++++--- 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index e47ad3a87b2..466a5c2dd4e 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -4,6 +4,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; import * as ManagedRelay from "../relay/managedRelay.ts"; import { remoteHttpClientLayer } from "../rpc/http.ts"; @@ -188,6 +189,48 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); + it.effect("revalidates a bearer descriptor after the cache expires", () => + Effect.gen(function* () { + const reassignedEnvironmentId = EnvironmentId.make("environment-2"); + const harness = yield* makeHarness({ + responses: [ + Response.json(DESCRIPTOR), + websocketTicket("first-ticket"), + Response.json({ + ...DESCRIPTOR, + environmentId: reassignedEnvironmentId, + }), + ], + }); + + const failure = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + const authorize = () => + remote.authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + bearerToken: "bearer-token", + }); + + yield* authorize(); + yield* TestClock.adjust("10 seconds"); + return yield* authorize().pipe(Effect.flip); + }).pipe(Effect.provide(Layer.merge(harness.layer, TestClock.layer()))); + + expect(failure).toEqual( + expect.objectContaining({ + _tag: "ConnectionBlockedError", + reason: "configuration", + detail: `Connected environment ${reassignedEnvironmentId} does not match ${ENVIRONMENT_ID}.`, + }), + ); + expect( + harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")), + ).toHaveLength(2); + }), + ); + it.effect("reuses a valid persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index ec517c98f56..410c7a39dc0 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -59,6 +59,7 @@ export class RemoteEnvironmentAuthorization extends Context.Service< const TOKEN_EXPIRY_SAFETY_MARGIN_MS = 60_000; const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; +const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000; function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) { return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError" @@ -85,6 +86,7 @@ export const make = Effect.gen(function* () { { readonly httpBaseUrl: string; readonly descriptor: ExecutionEnvironmentDescriptor; + readonly validatedAtEpochMs: number; } > >(new Map()); @@ -98,25 +100,29 @@ export const make = Effect.gen(function* () { readonly wsBaseUrl: string; readonly bearerToken: string; }) { + const now = yield* Clock.currentTimeMillis; const cachedDescriptor = (yield* Ref.get(bearerDescriptors)).get(input.expectedEnvironmentId); - const descriptor = - cachedDescriptor?.httpBaseUrl === input.httpBaseUrl - ? cachedDescriptor.descriptor - : yield* fetchDescriptor(input.httpBaseUrl).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const canReuseDescriptor = + cachedDescriptor?.httpBaseUrl === input.httpBaseUrl && + cachedDescriptor.validatedAtEpochMs + BEARER_DESCRIPTOR_CACHE_TTL_MS > now; + const descriptor = canReuseDescriptor + ? cachedDescriptor.descriptor + : yield* fetchDescriptor(input.httpBaseUrl).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, actual: descriptor.environmentId, }); } - if (cachedDescriptor?.httpBaseUrl !== input.httpBaseUrl) { + if (!canReuseDescriptor) { yield* Ref.update(bearerDescriptors, (current) => { const next = new Map(current); next.set(input.expectedEnvironmentId, { httpBaseUrl: input.httpBaseUrl, descriptor, + validatedAtEpochMs: now, }); return next; }); From d878362d8854618a7b7c7eb945273d65f2e7b558 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 30 Jul 2026 02:37:55 -0700 Subject: [PATCH 5/6] fix(mobile): reset retries on foreground reconnect --- .../src/connection/supervisor.test.ts | 56 ++++++++++++++++++ .../src/connection/supervisor.ts | 59 +++++++++++++------ 2 files changed, 98 insertions(+), 17 deletions(-) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 5d783996b3b..d305cb9d312 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -691,6 +691,62 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("restarts the retry ladder when a long resume replaces a connected session", () => + 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 && state.attempt === 2, + ); + + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 3 && state.attempt === 1, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("restarts the retry ladder when a long resume interrupts connection setup", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (attempt) => (attempt === 2 ? Effect.never : Effect.succeed(PREPARED_CONNECTION)), + }); + 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 === "connecting" && state.attempt === 2, + ); + + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1, + ); + }).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); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 61b824d2b1c..3b2ae49ab2c 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -64,6 +64,7 @@ type AttemptOutcome = readonly _tag: "Interrupted"; readonly established: boolean; readonly stable: boolean; + readonly resetRetry: boolean; } | { readonly _tag: "Failure"; @@ -83,7 +84,7 @@ type EstablishmentEvent = TracedAttemptFailure >; } - | { readonly _tag: "Interrupted" } + | { readonly _tag: "Interrupted"; readonly resetRetry: boolean } | { readonly _tag: "TimedOut" }; function exitUnlessInterrupted( @@ -169,7 +170,7 @@ function failureFromExit( stable: boolean, ): AttemptOutcome { if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) { - return { _tag: "Interrupted", established, stable }; + return { _tag: "Interrupted", established, stable, resetRetry: false }; } const typedFailure = exit.cause.reasons.find(Cause.isFailReason); if (typedFailure) { @@ -366,21 +367,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( switch (next._tag) { case "DisconnectRequested": case "RetryRequested": - return; + return false; case "NetworkChanged": if (next.network === "offline") { - return; + return false; } break; case "ConnectRequested": break; case "Wakeup": if (next.reason === "application-active-reconnect") { - return; + return true; } if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") { yield* logManagedRelayAccountChange; - return; + return false; } break; } @@ -395,22 +396,22 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( switch (next._tag) { case "DisconnectRequested": case "RetryRequested": - return; + return false; case "NetworkChanged": if (next.network === "offline") { - return; + return false; } break; case "Wakeup": if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") { yield* logManagedRelayAccountChange; - return; + return false; } if (next.reason === "application-active-reconnect") { // Mobile operating systems commonly suspend sockets without // delivering a close event. A long background resume deliberately // replaces that lease and starts a fresh attempt without backoff. - return; + return true; } if (next.reason === "application-active" || next.reason === "application-active-probe") { const probe = yield* lease.session.probe.pipe( @@ -446,21 +447,24 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( case "DisconnectRequested": case "RetryRequested": yield* Fiber.interrupt(probe); - return; + return false; case "NetworkChanged": if (probeEvent.signal.network === "offline") { yield* Fiber.interrupt(probe); - return; + return false; } break; case "Wakeup": + if (probeEvent.signal.reason === "application-active-reconnect") { + yield* Fiber.interrupt(probe); + return true; + } if ( - probeEvent.signal.reason === "application-active-reconnect" || - (probeEvent.signal.reason === "credentials-changed" && - target._tag === "RelayConnectionTarget") + probeEvent.signal.reason === "credentials-changed" && + target._tag === "RelayConnectionTarget" ) { yield* Fiber.interrupt(probe); - return; + return false; } break; case "ConnectRequested": @@ -493,7 +497,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }), ), ), - waitForEstablishmentInterrupt().pipe(Effect.as({ _tag: "Interrupted" })), + waitForEstablishmentInterrupt().pipe( + Effect.map( + (resetRetry): EstablishmentEvent => ({ + _tag: "Interrupted", + resetRetry, + }), + ), + ), Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe( Effect.as({ _tag: "TimedOut" }), ), @@ -504,6 +515,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( _tag: "Interrupted", established: false, stable: false, + resetRetry: establishment.resetRetry, } satisfies AttemptOutcome; } if (establishment._tag === "TimedOut") { @@ -546,6 +558,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( _tag: "Interrupted", established: false, stable: false, + resetRetry: false, } satisfies AttemptOutcome; } @@ -582,6 +595,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ), ).pipe(exitUnlessInterrupted); const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt; + if (Exit.isSuccess(connectedExit)) { + return { + _tag: "Interrupted", + established: true, + stable: connectedForMs >= BACKOFF_RESET_AFTER_MS, + resetRetry: connectedExit.value, + } satisfies AttemptOutcome; + } return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS); }, Effect.ensuring(clearLease)); @@ -645,6 +666,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } } if (outcome._tag === "Interrupted") { + if (outcome.resetRetry) { + failureCount = 0; + pendingRetry = Option.none(); + } continue; } From e27978741cbcf507fc96cc0fe909da8210b6420c Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 30 Jul 2026 04:00:21 -0700 Subject: [PATCH 6/6] fix(mobile): reset retries while offline --- .../src/connection/supervisor.test.ts | 64 +++++++++++++++++++ .../src/connection/supervisor.ts | 32 ++++++---- 2 files changed, 85 insertions(+), 11 deletions(-) diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index d305cb9d312..b31ea9b4fc9 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -311,6 +311,38 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("resets retries when activation arrives before the network returns", () => + 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* harness.setNetworkStatus("offline"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "offline" && state.attempt === 2, + ); + + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "offline" && state.attempt === 1, + ); + yield* harness.setNetworkStatus("online"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1, + ); + }), + ); + it.effect("retries forever with exponential backoff capped at sixteen seconds", () => Effect.gen(function* () { const harness = yield* makeHarness({ @@ -501,6 +533,38 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("resets retries when activation wakes a blocked connection", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: (attempt) => + attempt === 1 + ? Effect.fail(transient()) + : attempt === 2 + ? Effect.fail(blocked()) + : Effect.succeed(PREPARED_CONNECTION), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + yield* TestClock.adjust("1 second"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "blocked" && state.attempt === 2, + ); + + yield* harness.wake("application-active-reconnect"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.attempt === 1, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("releases a live session while offline and starts a new generation when online", () => Effect.gen(function* () { const harness = yield* makeHarness(); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 3b2ae49ab2c..e4ac359e5b1 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -626,20 +626,27 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ); }); - const waitForSignal = Queue.take(signals); + const waitForSignal = Queue.take(signals).pipe( + Effect.map( + (next) => next._tag === "Wakeup" && ConnectionWakeups.isApplicationActiveWakeup(next.reason), + ), + ); const run = Effect.fnUntraced(function* () { let failureCount = 0; let generation = 0; let latestFailure: ConnectionAttemptError | null = null; let pendingRetry = Option.none(); + const resetRetryLadder = () => { + failureCount = 0; + pendingRetry = Option.none(); + }; for (;;) { const currentIntent = yield* Ref.get(intent); if (!currentIntent.desired) { - failureCount = 0; + resetRetryLadder(); latestFailure = null; - pendingRetry = Option.none(); yield* clearLease; yield* setState(availableState(currentIntent, generation)); yield* waitForSignal; @@ -648,7 +655,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (currentIntent.network === "offline") { yield* clearLease; yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure)); - yield* waitForSignal; + const applicationActivated = yield* waitForSignal; + if (applicationActivated) { + resetRetryLadder(); + } continue; } @@ -660,15 +670,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if (outcome.established) { generation = nextGeneration; if (outcome.stable) { - failureCount = 0; + resetRetryLadder(); latestFailure = null; - pendingRetry = Option.none(); } } if (outcome._tag === "Interrupted") { if (outcome.resetRetry) { - failureCount = 0; - pendingRetry = Option.none(); + resetRetryLadder(); } continue; } @@ -688,7 +696,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( lastFailure: error, retryAt: null, }); - yield* waitForSignal; + const applicationActivated = yield* waitForSignal; + if (applicationActivated) { + resetRetryLadder(); + } continue; } @@ -713,8 +724,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }); const applicationActivated = yield* waitForRetrySignal(delayMs); if (applicationActivated) { - failureCount = 0; - pendingRetry = Option.none(); + resetRetryLadder(); } } });