authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-07-30 04:10:26-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-30 04:10:26-07:00
log69a18ce912ac3ae93cf498cfe0cee40beb49fb9f
treea9df8140e4ce993820d437c51386afec98bea00b
parentf0c6ba994670bbd9eeab97eab6701d5c3589178d
signature Signed by PGP key B5690EEEBB952194

perf(mobile): reconnect environments immediately on resume (#4878)

13 files changed, 547 insertions(+), 93 deletions(-)

apps/mobile/src/connection/app-state-wakeups.test.ts created+21
...@@ -0,0 +1,21 @@
1import { describe, expect, it } from "@effect/vitest";
2
3import {
4 MOBILE_BACKGROUND_RECONNECT_AFTER_MS,
5 mobileApplicationActiveWakeup,
6} from "./app-state-wakeups";
7
8describe("mobileApplicationActiveWakeup", () => {
9 it("uses a fast probe after a short interruption", () => {
10 expect(mobileApplicationActiveWakeup(null, 20_000)).toBe("application-active-probe");
11 expect(
12 mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS - 1),
13 ).toBe("application-active-probe");
14 });
15
16 it("replaces the session after a meaningful background suspension", () => {
17 expect(
18 mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS),
19 ).toBe("application-active-reconnect");
20 });
21});
apps/mobile/src/connection/app-state-wakeups.ts created+18
...@@ -0,0 +1,18 @@
1import type { Wakeups } from "@t3tools/client-runtime/connection";
2
3export const MOBILE_BACKGROUND_RECONNECT_AFTER_MS = 10_000;
4
5export type MobileApplicationActiveWakeup = Extract<
6 Wakeups.ConnectionWakeup,
7 "application-active-probe" | "application-active-reconnect"
8>;
9
10export function mobileApplicationActiveWakeup(
11 backgroundedAtMs: number | null,
12 activeAtMs: number,
13): MobileApplicationActiveWakeup {
14 return backgroundedAtMs !== null &&
15 activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS
16 ? "application-active-reconnect"
17 : "application-active-probe";
18}
apps/mobile/src/connection/platform.ts+38-11
...@@ -30,6 +30,7 @@ import * as MobileStorage from "../persistence/mobile-storage";...@@ -30,6 +30,7 @@ import * as MobileStorage from "../persistence/mobile-storage";
30import { appAtomRegistry } from "../state/atom-registry";30import { appAtomRegistry } from "../state/atom-registry";
31import { clearThreadOutboxEnvironment } from "../state/thread-outbox";31import { clearThreadOutboxEnvironment } from "../state/thread-outbox";
32import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts";32import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts";
33import { mobileApplicationActiveWakeup } from "./app-state-wakeups";
33import { connectionStorageLayer } from "./storage";34import { connectionStorageLayer } from "./storage";
3435
35function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" {36function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" {
...@@ -54,27 +55,53 @@ const connectivityLayer = Connectivity.layer({...@@ -54,27 +55,53 @@ const connectivityLayer = Connectivity.layer({
54 ),55 ),
55 changes: Stream.callback((queue) =>56 changes: Stream.callback((queue) =>
56 Effect.acquireRelease(57 Effect.acquireRelease(
57 Effect.sync(() =>58 Effect.sync(() => {
58 Network.addNetworkStateListener((state) => {59 let active = true;
60 const networkSubscription = Network.addNetworkStateListener((state) => {
59 Queue.offerUnsafe(queue, networkStatus(state));61 Queue.offerUnsafe(queue, networkStatus(state));
60 }),62 });
61 ),63 const appStateSubscription = AppState.addEventListener("change", (state) => {
62 (subscription) => Effect.sync(() => subscription.remove()),64 if (state !== "active") {
65 return;
66 }
67 void Network.getNetworkStateAsync()
68 .then((current) => {
69 if (active) {
70 Queue.offerUnsafe(queue, networkStatus(current));
71 }
72 })
73 .catch(() => undefined);
74 });
75 return {
76 close: () => {
77 active = false;
78 networkSubscription.remove();
79 appStateSubscription.remove();
80 },
81 };
82 }),
83 ({ close }) => Effect.sync(close),
63 ).pipe(Effect.asVoid),84 ).pipe(Effect.asVoid),
64 ),85 ),
65});86});
6687
67const wakeupsLayer = Wakeups.layer({88const wakeupsLayer = Wakeups.layer({
68 changes: Stream.merge(89 changes: Stream.merge(
69 Stream.callback<"application-active">((queue) =>90 Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) =>
70 Effect.acquireRelease(91 Effect.acquireRelease(
71 Effect.sync(() =>92 Effect.sync(() => {
72 AppState.addEventListener("change", (state) => {93 let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null;
94 return AppState.addEventListener("change", (state) => {
95 if (state === "background") {
96 backgroundedAtMs = Date.now();
97 return;
98 }
73 if (state === "active") {99 if (state === "active") {
74 Queue.offerUnsafe(queue, "application-active");100 Queue.offerUnsafe(queue, mobileApplicationActiveWakeup(backgroundedAtMs, Date.now()));
101 backgroundedAtMs = null;
75 }102 }
76 }),103 });
77 ),104 }),
78 (subscription) => Effect.sync(() => subscription.remove()),105 (subscription) => Effect.sync(() => subscription.remove()),
79 ).pipe(Effect.asVoid),106 ).pipe(Effect.asVoid),
80 ),107 ),
docs/architecture/connection-runtime.md+3
...@@ -40,6 +40,9 @@ The supervisor is the only retry owner....@@ -40,6 +40,9 @@ The supervisor is the only retry owner.
40 seconds.40 seconds.
415. Connectivity changes, application activation, credential changes, and415. Connectivity changes, application activation, credential changes, and
42 explicit user retry interrupt the current wait and trigger a fresh attempt.42 explicit user retry interrupt the current wait and trigger a fresh attempt.
43 Application activation also resets the backoff ladder. Mobile briefly probes
44 a session after short interruptions and replaces it immediately after a
45 meaningful background suspension.
436. Authentication or configuration failures remain blocked until an external466. Authentication or configuration failures remain blocked until an external
44 wakeup changes the relevant input.47 wakeup changes the relevant input.
457. An involuntary session close keeps the registration and cache, then retries.487. An involuntary session close keeps the registration and cache, then retries.
packages/client-runtime/src/authorization/layer.test.ts+78-14
...@@ -4,6 +4,7 @@ import * as Effect from "effect/Effect";...@@ -4,6 +4,7 @@ import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";5import * as Option from "effect/Option";
6import * as Ref from "effect/Ref";6import * as Ref from "effect/Ref";
7import * as TestClock from "effect/testing/TestClock";
78
8import * as ManagedRelay from "../relay/managedRelay.ts";9import * as ManagedRelay from "../relay/managedRelay.ts";
9import { remoteHttpClientLayer } from "../rpc/http.ts";10import { remoteHttpClientLayer } from "../rpc/http.ts";
...@@ -155,6 +156,81 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* (...@@ -155,6 +156,81 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* (
155});156});
156157
157describe("RemoteEnvironmentAuthorization", () => {158describe("RemoteEnvironmentAuthorization", () => {
159 it.effect("reuses a validated bearer descriptor while issuing fresh websocket tickets", () =>
160 Effect.gen(function* () {
161 const harness = yield* makeHarness({
162 responses: [
163 Response.json(DESCRIPTOR),
164 websocketTicket("first-ticket"),
165 websocketTicket("second-ticket"),
166 ],
167 });
168
169 const [first, second] = yield* Effect.gen(function* () {
170 const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
171 const authorize = () =>
172 remote.authorizeBearer({
173 expectedEnvironmentId: ENVIRONMENT_ID,
174 httpBaseUrl: ENDPOINT.httpBaseUrl,
175 wsBaseUrl: ENDPOINT.wsBaseUrl,
176 bearerToken: "bearer-token",
177 });
178 return [yield* authorize(), yield* authorize()] as const;
179 }).pipe(Effect.provide(harness.layer));
180
181 expect(first.socketUrl).toContain("wsTicket=first-ticket");
182 expect(second.socketUrl).toContain("wsTicket=second-ticket");
183 expect(
184 harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")),
185 ).toHaveLength(1);
186 expect(
187 harness.fetch.calls.filter(([url]) => String(url).endsWith("/api/auth/websocket-ticket")),
188 ).toHaveLength(2);
189 }),
190 );
191
192 it.effect("revalidates a bearer descriptor after the cache expires", () =>
193 Effect.gen(function* () {
194 const reassignedEnvironmentId = EnvironmentId.make("environment-2");
195 const harness = yield* makeHarness({
196 responses: [
197 Response.json(DESCRIPTOR),
198 websocketTicket("first-ticket"),
199 Response.json({
200 ...DESCRIPTOR,
201 environmentId: reassignedEnvironmentId,
202 }),
203 ],
204 });
205
206 const failure = yield* Effect.gen(function* () {
207 const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
208 const authorize = () =>
209 remote.authorizeBearer({
210 expectedEnvironmentId: ENVIRONMENT_ID,
211 httpBaseUrl: ENDPOINT.httpBaseUrl,
212 wsBaseUrl: ENDPOINT.wsBaseUrl,
213 bearerToken: "bearer-token",
214 });
215
216 yield* authorize();
217 yield* TestClock.adjust("10 seconds");
218 return yield* authorize().pipe(Effect.flip);
219 }).pipe(Effect.provide(Layer.merge(harness.layer, TestClock.layer())));
220
221 expect(failure).toEqual(
222 expect.objectContaining({
223 _tag: "ConnectionBlockedError",
224 reason: "configuration",
225 detail: `Connected environment ${reassignedEnvironmentId} does not match ${ENVIRONMENT_ID}.`,
226 }),
227 );
228 expect(
229 harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")),
230 ).toHaveLength(2);
231 }),
232 );
233
158 it.effect("reuses a valid persisted environment token without contacting the relay", () =>234 it.effect("reuses a valid persisted environment token without contacting the relay", () =>
159 Effect.gen(function* () {235 Effect.gen(function* () {
160 const cached = new TokenStore.RemoteDpopAccessToken({236 const cached = new TokenStore.RemoteDpopAccessToken({
...@@ -265,7 +341,7 @@ describe("RemoteEnvironmentAuthorization", () => {...@@ -265,7 +341,7 @@ describe("RemoteEnvironmentAuthorization", () => {
265 }),341 }),
266 );342 );
267343
268 it.effect("refreshes a cached endpoint after consecutive transient failures", () =>344 it.effect("refreshes a cached endpoint after its first transient failure", () =>
269 Effect.gen(function* () {345 Effect.gen(function* () {
270 const cached = new TokenStore.RemoteDpopAccessToken({346 const cached = new TokenStore.RemoteDpopAccessToken({
271 environmentId: ENVIRONMENT_ID,347 environmentId: ENVIRONMENT_ID,
...@@ -279,7 +355,6 @@ describe("RemoteEnvironmentAuthorization", () => {...@@ -279,7 +355,6 @@ describe("RemoteEnvironmentAuthorization", () => {
279 initialToken: cached,355 initialToken: cached,
280 responses: [356 responses: [
281 new Response("endpoint unavailable", { status: 503 }),357 new Response("endpoint unavailable", { status: 503 }),
282 new Response("endpoint still unavailable", { status: 503 }),
283 Response.json(DESCRIPTOR),358 Response.json(DESCRIPTOR),
284 accessToken("replacement-access-token"),359 accessToken("replacement-access-token"),
285 websocketTicket("replacement-ticket"),360 websocketTicket("replacement-ticket"),
...@@ -288,17 +363,6 @@ describe("RemoteEnvironmentAuthorization", () => {...@@ -288,17 +363,6 @@ describe("RemoteEnvironmentAuthorization", () => {
288363
289 const authorized = yield* Effect.gen(function* () {364 const authorized = yield* Effect.gen(function* () {
290 const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;365 const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
291 const firstFailure = yield* remote
292 .authorizeDpop({
293 expectedEnvironmentId: ENVIRONMENT_ID,
294 obtainBootstrap: harness.obtainBootstrap,
295 })
296 .pipe(Effect.flip);
297
298 expect(firstFailure._tag).toBe("ConnectionTransientError");
299 expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0);
300 expect((yield* Ref.get(harness.tokens)).get(ENVIRONMENT_ID)).toBe(cached);
301
302 return yield* remote.authorizeDpop({366 return yield* remote.authorizeDpop({
303 expectedEnvironmentId: ENVIRONMENT_ID,367 expectedEnvironmentId: ENVIRONMENT_ID,
304 obtainBootstrap: harness.obtainBootstrap,368 obtainBootstrap: harness.obtainBootstrap,
...@@ -312,7 +376,7 @@ describe("RemoteEnvironmentAuthorization", () => {...@@ -312,7 +376,7 @@ describe("RemoteEnvironmentAuthorization", () => {
312 accessToken: "replacement-access-token",376 accessToken: "replacement-access-token",
313 }),377 }),
314 );378 );
315 expect(harness.fetch.calls).toHaveLength(5);379 expect(harness.fetch.calls).toHaveLength(4);
316 }),380 }),
317 );381 );
318382
packages/client-runtime/src/authorization/service.ts+39-35
...@@ -1,4 +1,4 @@...@@ -1,4 +1,4 @@
1import { EnvironmentId } from "@t3tools/contracts";1import { EnvironmentId, type ExecutionEnvironmentDescriptor } from "@t3tools/contracts";
2import type { RelayManagedEndpoint } from "@t3tools/contracts/relay";2import type { RelayManagedEndpoint } from "@t3tools/contracts/relay";
3import {3import {
4 exchangeRemoteDpopAccessToken,4 exchangeRemoteDpopAccessToken,
...@@ -58,7 +58,8 @@ export class RemoteEnvironmentAuthorization extends Context.Service<...@@ -58,7 +58,8 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
58>()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {}58>()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {}
5959
60const TOKEN_EXPIRY_SAFETY_MARGIN_MS = 60_000;60const TOKEN_EXPIRY_SAFETY_MARGIN_MS = 60_000;
61const CACHED_ENDPOINT_FAILURE_THRESHOLD = 2;61const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000;
62const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000;
6263
63function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) {64function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) {
64 return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError"65 return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError"
...@@ -79,25 +80,16 @@ export const make = Effect.gen(function* () {...@@ -79,25 +80,16 @@ export const make = Effect.gen(function* () {
79 const presentation = yield* ClientCapabilities.ClientPresentation;80 const presentation = yield* ClientCapabilities.ClientPresentation;
80 const tokenStore = yield* TokenStore.RemoteDpopAccessTokenStore;81 const tokenStore = yield* TokenStore.RemoteDpopAccessTokenStore;
81 const httpClient = yield* HttpClient.HttpClient;82 const httpClient = yield* HttpClient.HttpClient;
82 const cachedEndpointFailures = yield* Ref.make<ReadonlyMap<string, number>>(new Map());83 const bearerDescriptors = yield* Ref.make<
8384 ReadonlyMap<
84 const resetCachedEndpointFailures = (environmentId: string) =>85 EnvironmentId,
85 Ref.update(cachedEndpointFailures, (current) => {86 {
86 if (!current.has(environmentId)) {87 readonly httpBaseUrl: string;
87 return current;88 readonly descriptor: ExecutionEnvironmentDescriptor;
89 readonly validatedAtEpochMs: number;
88 }90 }
89 const next = new Map(current);91 >
90 next.delete(environmentId);92 >(new Map());
91 return next;
92 });
93
94 const recordCachedEndpointFailure = (environmentId: string) =>
95 Ref.modify(cachedEndpointFailures, (current) => {
96 const failureCount = (current.get(environmentId) ?? 0) + 1;
97 const next = new Map(current);
98 next.set(environmentId, failureCount);
99 return [failureCount, next] as const;
100 });
10193
102 const authorizeBearer = Effect.fn("clientRuntime.connection.remote.authorizeBearer")(94 const authorizeBearer = Effect.fn("clientRuntime.connection.remote.authorizeBearer")(
103 function* (input: {95 function* (input: {
...@@ -108,15 +100,33 @@ export const make = Effect.gen(function* () {...@@ -108,15 +100,33 @@ export const make = Effect.gen(function* () {
108 readonly wsBaseUrl: string;100 readonly wsBaseUrl: string;
109 readonly bearerToken: string;101 readonly bearerToken: string;
110 }) {102 }) {
111 const descriptor = yield* fetchDescriptor(input.httpBaseUrl).pipe(103 const now = yield* Clock.currentTimeMillis;
112 Effect.provideService(HttpClient.HttpClient, httpClient),104 const cachedDescriptor = (yield* Ref.get(bearerDescriptors)).get(input.expectedEnvironmentId);
113 );105 const canReuseDescriptor =
106 cachedDescriptor?.httpBaseUrl === input.httpBaseUrl &&
107 cachedDescriptor.validatedAtEpochMs + BEARER_DESCRIPTOR_CACHE_TTL_MS > now;
108 const descriptor = canReuseDescriptor
109 ? cachedDescriptor.descriptor
110 : yield* fetchDescriptor(input.httpBaseUrl).pipe(
111 Effect.provideService(HttpClient.HttpClient, httpClient),
112 );
114 if (descriptor.environmentId !== input.expectedEnvironmentId) {113 if (descriptor.environmentId !== input.expectedEnvironmentId) {
115 return yield* environmentMismatchError({114 return yield* environmentMismatchError({
116 expected: input.expectedEnvironmentId,115 expected: input.expectedEnvironmentId,
117 actual: descriptor.environmentId,116 actual: descriptor.environmentId,
118 });117 });
119 }118 }
119 if (!canReuseDescriptor) {
120 yield* Ref.update(bearerDescriptors, (current) => {
121 const next = new Map(current);
122 next.set(input.expectedEnvironmentId, {
123 httpBaseUrl: input.httpBaseUrl,
124 descriptor,
125 validatedAtEpochMs: now,
126 });
127 return next;
128 });
129 }
120 const socketUrl = yield* resolveRemoteWebSocketConnectionUrl({130 const socketUrl = yield* resolveRemoteWebSocketConnectionUrl({
121 wsBaseUrl: input.wsBaseUrl,131 wsBaseUrl: input.wsBaseUrl,
122 httpBaseUrl: input.httpBaseUrl,132 httpBaseUrl: input.httpBaseUrl,
...@@ -139,7 +149,7 @@ export const make = Effect.gen(function* () {...@@ -139,7 +149,7 @@ export const make = Effect.gen(function* () {
139 );149 );
140150
141 const createDpopSocketUrl = Effect.fn("clientRuntime.connection.remote.createDpopSocketUrl")(151 const createDpopSocketUrl = Effect.fn("clientRuntime.connection.remote.createDpopSocketUrl")(
142 function* (token: TokenStore.RemoteDpopAccessToken) {152 function* (token: TokenStore.RemoteDpopAccessToken, timeoutMs?: number) {
143 const ticketProof = yield* signer153 const ticketProof = yield* signer
144 .createProof({154 .createProof({
145 method: "POST",155 method: "POST",
...@@ -160,6 +170,7 @@ export const make = Effect.gen(function* () {...@@ -160,6 +170,7 @@ export const make = Effect.gen(function* () {
160 httpBaseUrl: token.endpoint.httpBaseUrl,170 httpBaseUrl: token.endpoint.httpBaseUrl,
161 accessToken: token.accessToken,171 accessToken: token.accessToken,
162 dpopProof: ticketProof,172 dpopProof: ticketProof,
173 ...(timeoutMs === undefined ? {} : { timeoutMs }),
163 }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient));174 }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient));
164 },175 },
165 );176 );
...@@ -196,9 +207,11 @@ export const make = Effect.gen(function* () {...@@ -196,9 +207,11 @@ export const make = Effect.gen(function* () {
196 yield* Effect.annotateCurrentSpan({207 yield* Effect.annotateCurrentSpan({
197 "connection.remote_token_cache": "hit",208 "connection.remote_token_cache": "hit",
198 });209 });
199 const cachedSocket = yield* createDpopSocketUrl(cached.value).pipe(Effect.result);210 const cachedSocket = yield* createDpopSocketUrl(
211 cached.value,
212 CACHED_ENDPOINT_SOCKET_TIMEOUT_MS,
213 ).pipe(Effect.result);
200 if (Result.isSuccess(cachedSocket)) {214 if (Result.isSuccess(cachedSocket)) {
201 yield* resetCachedEndpointFailures(input.expectedEnvironmentId);
202 return {215 return {
203 environmentId: cached.value.environmentId,216 environmentId: cached.value.environmentId,
204 label: cached.value.label,217 label: cached.value.label,
...@@ -213,20 +226,11 @@ export const make = Effect.gen(function* () {...@@ -213,20 +226,11 @@ export const make = Effect.gen(function* () {
213 if (cachedSocket.failure._tag === "ConnectionBlockedError") {226 if (cachedSocket.failure._tag === "ConnectionBlockedError") {
214 return yield* mapDpopSocketError(cachedSocket.failure);227 return yield* mapDpopSocketError(cachedSocket.failure);
215 }228 }
216 const mappedFailure = mapDpopSocketError(cachedSocket.failure);
217 if (mappedFailure._tag === "ConnectionTransientError") {
218 const failureCount = yield* recordCachedEndpointFailure(input.expectedEnvironmentId);
219 if (failureCount < CACHED_ENDPOINT_FAILURE_THRESHOLD) {
220 return yield* mappedFailure;
221 }
222 }
223 yield* tokenStore229 yield* tokenStore
224 .remove(input.expectedEnvironmentId)230 .remove(input.expectedEnvironmentId)
225 .pipe(Effect.withSpan("environment.authorization.accessToken.remove"));231 .pipe(Effect.withSpan("environment.authorization.accessToken.remove"));
226 yield* resetCachedEndpointFailures(input.expectedEnvironmentId);
227 }232 }
228233
229 yield* resetCachedEndpointFailures(input.expectedEnvironmentId);
230 yield* Effect.annotateCurrentSpan({234 yield* Effect.annotateCurrentSpan({
231 "connection.remote_token_cache": "miss",235 "connection.remote_token_cache": "miss",
232 });236 });
packages/client-runtime/src/connection/supervisor.test.ts+216-5
...@@ -124,7 +124,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:...@@ -124,7 +124,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
124 const releaseCount = yield* Ref.make(0);124 const releaseCount = yield* Ref.make(0);
125 const wakeups = yield* SubscriptionRef.make<{125 const wakeups = yield* SubscriptionRef.make<{
126 readonly sequence: number;126 readonly sequence: number;
127 readonly reason: "application-active" | "credentials-changed";127 readonly reason: ConnectionWakeups.ConnectionWakeup;
128 }>({128 }>({
129 sequence: 0,129 sequence: 0,
130 reason: "application-active",130 reason: "application-active",
...@@ -198,7 +198,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:...@@ -198,7 +198,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
198 sessionCount,198 sessionCount,
199 releaseCount,199 releaseCount,
200 setNetworkStatus: (status: NetworkStatus) => SubscriptionRef.set(networkStatus, status),200 setNetworkStatus: (status: NetworkStatus) => SubscriptionRef.set(networkStatus, status),
201 wake: (reason: "application-active" | "credentials-changed") =>201 wake: (reason: ConnectionWakeups.ConnectionWakeup) =>
202 SubscriptionRef.update(wakeups, (event) => ({202 SubscriptionRef.update(wakeups, (event) => ({
203 sequence: event.sequence + 1,203 sequence: event.sequence + 1,
204 reason,204 reason,
...@@ -311,6 +311,38 @@ describe("EnvironmentSupervisor", () => {...@@ -311,6 +311,38 @@ describe("EnvironmentSupervisor", () => {
311 }),311 }),
312 );312 );
313313
314 it.effect("resets retries when activation arrives before the network returns", () =>
315 Effect.gen(function* () {
316 const harness = yield* makeHarness();
317 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
318 initiallyDesired: true,
319 }).pipe(Effect.provide(harness.dependencies));
320
321 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
322 yield* harness.closeLatestSession();
323 yield* awaitState(
324 supervisor.state,
325 (state) => state.phase === "backoff" && state.attempt === 1,
326 );
327 yield* harness.setNetworkStatus("offline");
328 yield* awaitState(
329 supervisor.state,
330 (state) => state.phase === "offline" && state.attempt === 2,
331 );
332
333 yield* harness.wake("application-active-reconnect");
334 yield* awaitState(
335 supervisor.state,
336 (state) => state.phase === "offline" && state.attempt === 1,
337 );
338 yield* harness.setNetworkStatus("online");
339 yield* awaitState(
340 supervisor.state,
341 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1,
342 );
343 }),
344 );
345
314 it.effect("retries forever with exponential backoff capped at sixteen seconds", () =>346 it.effect("retries forever with exponential backoff capped at sixteen seconds", () =>
315 Effect.gen(function* () {347 Effect.gen(function* () {
316 const harness = yield* makeHarness({348 const harness = yield* makeHarness({
...@@ -501,6 +533,38 @@ describe("EnvironmentSupervisor", () => {...@@ -501,6 +533,38 @@ describe("EnvironmentSupervisor", () => {
501 }).pipe(Effect.provide(TestClock.layer())),533 }).pipe(Effect.provide(TestClock.layer())),
502 );534 );
503535
536 it.effect("resets retries when activation wakes a blocked connection", () =>
537 Effect.gen(function* () {
538 const harness = yield* makeHarness({
539 prepare: (attempt) =>
540 attempt === 1
541 ? Effect.fail(transient())
542 : attempt === 2
543 ? Effect.fail(blocked())
544 : Effect.succeed(PREPARED_CONNECTION),
545 });
546 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
547 initiallyDesired: true,
548 }).pipe(Effect.provide(harness.dependencies));
549
550 yield* awaitState(
551 supervisor.state,
552 (state) => state.phase === "backoff" && state.attempt === 1,
553 );
554 yield* TestClock.adjust("1 second");
555 yield* awaitState(
556 supervisor.state,
557 (state) => state.phase === "blocked" && state.attempt === 2,
558 );
559
560 yield* harness.wake("application-active-reconnect");
561 yield* awaitState(
562 supervisor.state,
563 (state) => state.phase === "connected" && state.attempt === 1,
564 );
565 }).pipe(Effect.provide(TestClock.layer())),
566 );
567
504 it.effect("releases a live session while offline and starts a new generation when online", () =>568 it.effect("releases a live session while offline and starts a new generation when online", () =>
505 Effect.gen(function* () {569 Effect.gen(function* () {
506 const harness = yield* makeHarness();570 const harness = yield* makeHarness();
...@@ -652,6 +716,101 @@ describe("EnvironmentSupervisor", () => {...@@ -652,6 +716,101 @@ describe("EnvironmentSupervisor", () => {
652 }).pipe(Effect.provide(TestClock.layer())),716 }).pipe(Effect.provide(TestClock.layer())),
653 );717 );
654718
719 it.effect("restarts the retry ladder when mobile returns to the foreground", () =>
720 Effect.gen(function* () {
721 const harness = yield* makeHarness();
722 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
723 initiallyDesired: true,
724 }).pipe(Effect.provide(harness.dependencies));
725
726 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
727 yield* harness.closeLatestSession();
728 yield* awaitState(
729 supervisor.state,
730 (state) => state.phase === "backoff" && state.attempt === 1,
731 );
732 yield* TestClock.adjust("1 second");
733 yield* awaitState(
734 supervisor.state,
735 (state) => state.phase === "connected" && state.generation === 2,
736 );
737 yield* harness.closeLatestSession();
738 yield* awaitState(
739 supervisor.state,
740 (state) => state.phase === "backoff" && state.attempt === 2,
741 );
742
743 yield* harness.wake("application-active-reconnect");
744 yield* awaitState(
745 supervisor.state,
746 (state) => state.phase === "connected" && state.generation === 3 && state.attempt === 1,
747 );
748 yield* harness.closeLatestSession();
749 yield* awaitState(
750 supervisor.state,
751 (state) => state.phase === "backoff" && state.attempt === 1,
752 );
753
754 expect(yield* Ref.get(harness.sessionCount)).toBe(3);
755 }).pipe(Effect.provide(TestClock.layer())),
756 );
757
758 it.effect("restarts the retry ladder when a long resume replaces a connected session", () =>
759 Effect.gen(function* () {
760 const harness = yield* makeHarness();
761 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
762 initiallyDesired: true,
763 }).pipe(Effect.provide(harness.dependencies));
764
765 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
766 yield* harness.closeLatestSession();
767 yield* awaitState(
768 supervisor.state,
769 (state) => state.phase === "backoff" && state.attempt === 1,
770 );
771 yield* TestClock.adjust("1 second");
772 yield* awaitState(
773 supervisor.state,
774 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 2,
775 );
776
777 yield* harness.wake("application-active-reconnect");
778 yield* awaitState(
779 supervisor.state,
780 (state) => state.phase === "connected" && state.generation === 3 && state.attempt === 1,
781 );
782 }).pipe(Effect.provide(TestClock.layer())),
783 );
784
785 it.effect("restarts the retry ladder when a long resume interrupts connection setup", () =>
786 Effect.gen(function* () {
787 const harness = yield* makeHarness({
788 prepare: (attempt) => (attempt === 2 ? Effect.never : Effect.succeed(PREPARED_CONNECTION)),
789 });
790 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
791 initiallyDesired: true,
792 }).pipe(Effect.provide(harness.dependencies));
793
794 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
795 yield* harness.closeLatestSession();
796 yield* awaitState(
797 supervisor.state,
798 (state) => state.phase === "backoff" && state.attempt === 1,
799 );
800 yield* TestClock.adjust("1 second");
801 yield* awaitState(
802 supervisor.state,
803 (state) => state.phase === "connecting" && state.attempt === 2,
804 );
805
806 yield* harness.wake("application-active-reconnect");
807 yield* awaitState(
808 supervisor.state,
809 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1,
810 );
811 }).pipe(Effect.provide(TestClock.layer())),
812 );
813
655 it.effect("probes the active session without reconnecting on application activation", () =>814 it.effect("probes the active session without reconnecting on application activation", () =>
656 Effect.gen(function* () {815 Effect.gen(function* () {
657 const probeCount = yield* Ref.make(0);816 const probeCount = yield* Ref.make(0);
...@@ -677,6 +836,58 @@ describe("EnvironmentSupervisor", () => {...@@ -677,6 +836,58 @@ describe("EnvironmentSupervisor", () => {
677 }),836 }),
678 );837 );
679838
839 it.effect("immediately replaces a mobile session after a long background resume", () =>
840 Effect.gen(function* () {
841 const probeCount = yield* Ref.make(0);
842 const harness = yield* makeHarness({
843 probe: () => Ref.update(probeCount, (count) => count + 1),
844 });
845 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
846 initiallyDesired: true,
847 }).pipe(Effect.provide(harness.dependencies));
848
849 yield* awaitState(
850 supervisor.state,
851 (state) => state.phase === "connected" && state.generation === 1,
852 );
853 yield* harness.wake("application-active-reconnect");
854 yield* awaitState(
855 supervisor.state,
856 (state) => state.phase === "connected" && state.generation === 2,
857 );
858
859 expect(yield* Ref.get(probeCount)).toBe(0);
860 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
861 expect(yield* Ref.get(harness.releaseCount)).toBe(1);
862 }),
863 );
864
865 it.effect("replaces a mobile session when a long resume interrupts an active probe", () =>
866 Effect.gen(function* () {
867 const harness = yield* makeHarness({
868 probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
869 });
870 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
871 initiallyDesired: true,
872 }).pipe(Effect.provide(harness.dependencies));
873
874 yield* awaitState(
875 supervisor.state,
876 (state) => state.phase === "connected" && state.generation === 1,
877 );
878 yield* harness.wake("application-active-probe");
879 yield* Effect.yieldNow;
880 yield* harness.wake("application-active-reconnect");
881 yield* awaitState(
882 supervisor.state,
883 (state) => state.phase === "connected" && state.generation === 2,
884 );
885
886 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
887 expect(yield* Ref.get(harness.releaseCount)).toBe(1);
888 }),
889 );
890
680 it.effect("reconnects when the foreground liveness probe fails", () =>891 it.effect("reconnects when the foreground liveness probe fails", () =>
681 Effect.gen(function* () {892 Effect.gen(function* () {
682 const harness = yield* makeHarness({893 const harness = yield* makeHarness({
...@@ -701,7 +912,7 @@ describe("EnvironmentSupervisor", () => {...@@ -701,7 +912,7 @@ describe("EnvironmentSupervisor", () => {
701 }).pipe(Effect.provide(TestClock.layer())),912 }).pipe(Effect.provide(TestClock.layer())),
702 );913 );
703914
704 it.effect("times out a stalled foreground liveness probe and reconnects", () =>915 it.effect("quickly times out a stalled mobile foreground liveness probe", () =>
705 Effect.gen(function* () {916 Effect.gen(function* () {
706 const harness = yield* makeHarness({917 const harness = yield* makeHarness({
707 probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),918 probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
...@@ -711,8 +922,8 @@ describe("EnvironmentSupervisor", () => {...@@ -711,8 +922,8 @@ describe("EnvironmentSupervisor", () => {
711 }).pipe(Effect.provide(harness.dependencies));922 }).pipe(Effect.provide(harness.dependencies));
712923
713 yield* awaitState(supervisor.state, (state) => state.phase === "connected");924 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
714 yield* harness.wake("application-active");925 yield* harness.wake("application-active-probe");
715 yield* TestClock.adjust("15 seconds");926 yield* TestClock.adjust("3 seconds");
716 yield* awaitState(927 yield* awaitState(
717 supervisor.state,928 supervisor.state,
718 (state) => state.phase === "backoff" && state.lastFailure?.reason === "timeout",929 (state) => state.phase === "backoff" && state.lastFailure?.reason === "timeout",
packages/client-runtime/src/connection/supervisor.ts+87-25
...@@ -32,6 +32,7 @@ import * as ConnectionWakeups from "./wakeups.ts";...@@ -32,6 +32,7 @@ import * as ConnectionWakeups from "./wakeups.ts";
32const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const;32const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const;
33const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";33const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";
34const CONNECTION_PROBE_TIMEOUT = "15 seconds";34const CONNECTION_PROBE_TIMEOUT = "15 seconds";
35const MOBILE_CONNECTION_PROBE_TIMEOUT = "3 seconds";
35const BACKOFF_RESET_AFTER_MS = 30_000;36const BACKOFF_RESET_AFTER_MS = 30_000;
3637
37interface SupervisorIntent {38interface SupervisorIntent {
...@@ -63,6 +64,7 @@ type AttemptOutcome =...@@ -63,6 +64,7 @@ type AttemptOutcome =
63 readonly _tag: "Interrupted";64 readonly _tag: "Interrupted";
64 readonly established: boolean;65 readonly established: boolean;
65 readonly stable: boolean;66 readonly stable: boolean;
67 readonly resetRetry: boolean;
66 }68 }
67 | {69 | {
68 readonly _tag: "Failure";70 readonly _tag: "Failure";
...@@ -82,7 +84,7 @@ type EstablishmentEvent =...@@ -82,7 +84,7 @@ type EstablishmentEvent =
82 TracedAttemptFailure84 TracedAttemptFailure
83 >;85 >;
84 }86 }
85 | { readonly _tag: "Interrupted" }87 | { readonly _tag: "Interrupted"; readonly resetRetry: boolean }
86 | { readonly _tag: "TimedOut" };88 | { readonly _tag: "TimedOut" };
8789
88function exitUnlessInterrupted<A, E, R>(90function exitUnlessInterrupted<A, E, R>(
...@@ -168,7 +170,7 @@ function failureFromExit<A>(...@@ -168,7 +170,7 @@ function failureFromExit<A>(
168 stable: boolean,170 stable: boolean,
169): AttemptOutcome {171): AttemptOutcome {
170 if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) {172 if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) {
171 return { _tag: "Interrupted", established, stable };173 return { _tag: "Interrupted", established, stable, resetRetry: false };
172 }174 }
173 const typedFailure = exit.cause.reasons.find(Cause.isFailReason);175 const typedFailure = exit.cause.reasons.find(Cause.isFailReason);
174 if (typedFailure) {176 if (typedFailure) {
...@@ -365,18 +367,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -365,18 +367,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
365 switch (next._tag) {367 switch (next._tag) {
366 case "DisconnectRequested":368 case "DisconnectRequested":
367 case "RetryRequested":369 case "RetryRequested":
368 return;370 return false;
369 case "NetworkChanged":371 case "NetworkChanged":
370 if (next.network === "offline") {372 if (next.network === "offline") {
371 return;373 return false;
372 }374 }
373 break;375 break;
374 case "ConnectRequested":376 case "ConnectRequested":
375 break;377 break;
376 case "Wakeup":378 case "Wakeup":
379 if (next.reason === "application-active-reconnect") {
380 return true;
381 }
377 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {382 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {
378 yield* logManagedRelayAccountChange;383 yield* logManagedRelayAccountChange;
379 return;384 return false;
380 }385 }
381 break;386 break;
382 }387 }
...@@ -391,21 +396,30 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -391,21 +396,30 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
391 switch (next._tag) {396 switch (next._tag) {
392 case "DisconnectRequested":397 case "DisconnectRequested":
393 case "RetryRequested":398 case "RetryRequested":
394 return;399 return false;
395 case "NetworkChanged":400 case "NetworkChanged":
396 if (next.network === "offline") {401 if (next.network === "offline") {
397 return;402 return false;
398 }403 }
399 break;404 break;
400 case "Wakeup":405 case "Wakeup":
401 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {406 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {
402 yield* logManagedRelayAccountChange;407 yield* logManagedRelayAccountChange;
403 return;408 return false;
409 }
410 if (next.reason === "application-active-reconnect") {
411 // Mobile operating systems commonly suspend sockets without
412 // delivering a close event. A long background resume deliberately
413 // replaces that lease and starts a fresh attempt without backoff.
414 return true;
404 }415 }
405 if (next.reason === "application-active") {416 if (next.reason === "application-active" || next.reason === "application-active-probe") {
406 const probe = yield* lease.session.probe.pipe(417 const probe = yield* lease.session.probe.pipe(
407 Effect.timeoutOrElse({418 Effect.timeoutOrElse({
408 duration: CONNECTION_PROBE_TIMEOUT,419 duration:
420 next.reason === "application-active-probe"
421 ? MOBILE_CONNECTION_PROBE_TIMEOUT
422 : CONNECTION_PROBE_TIMEOUT,
409 orElse: () =>423 orElse: () =>
410 Effect.fail(424 Effect.fail(
411 new ConnectionTransientError({425 new ConnectionTransientError({
...@@ -433,15 +447,27 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -433,15 +447,27 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
433 case "DisconnectRequested":447 case "DisconnectRequested":
434 case "RetryRequested":448 case "RetryRequested":
435 yield* Fiber.interrupt(probe);449 yield* Fiber.interrupt(probe);
436 return;450 return false;
437 case "NetworkChanged":451 case "NetworkChanged":
438 if (probeEvent.signal.network === "offline") {452 if (probeEvent.signal.network === "offline") {
439 yield* Fiber.interrupt(probe);453 yield* Fiber.interrupt(probe);
440 return;454 return false;
441 }455 }
442 break;456 break;
443 case "ConnectRequested":
444 case "Wakeup":457 case "Wakeup":
458 if (probeEvent.signal.reason === "application-active-reconnect") {
459 yield* Fiber.interrupt(probe);
460 return true;
461 }
462 if (
463 probeEvent.signal.reason === "credentials-changed" &&
464 target._tag === "RelayConnectionTarget"
465 ) {
466 yield* Fiber.interrupt(probe);
467 return false;
468 }
469 break;
470 case "ConnectRequested":
445 break;471 break;
446 }472 }
447 }473 }
...@@ -471,7 +497,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -471,7 +497,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
471 }),497 }),
472 ),498 ),
473 ),499 ),
474 waitForEstablishmentInterrupt().pipe(Effect.as<EstablishmentEvent>({ _tag: "Interrupted" })),500 waitForEstablishmentInterrupt().pipe(
501 Effect.map(
502 (resetRetry): EstablishmentEvent => ({
503 _tag: "Interrupted",
504 resetRetry,
505 }),
506 ),
507 ),
475 Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe(508 Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe(
476 Effect.as<EstablishmentEvent>({ _tag: "TimedOut" }),509 Effect.as<EstablishmentEvent>({ _tag: "TimedOut" }),
477 ),510 ),
...@@ -482,6 +515,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -482,6 +515,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
482 _tag: "Interrupted",515 _tag: "Interrupted",
483 established: false,516 established: false,
484 stable: false,517 stable: false,
518 resetRetry: establishment.resetRetry,
485 } satisfies AttemptOutcome;519 } satisfies AttemptOutcome;
486 }520 }
487 if (establishment._tag === "TimedOut") {521 if (establishment._tag === "TimedOut") {
...@@ -524,6 +558,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -524,6 +558,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
524 _tag: "Interrupted",558 _tag: "Interrupted",
525 established: false,559 established: false,
526 stable: false,560 stable: false,
561 resetRetry: false,
527 } satisfies AttemptOutcome;562 } satisfies AttemptOutcome;
528 }563 }
529564
...@@ -560,42 +595,58 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -560,42 +595,58 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
560 ),595 ),
561 ).pipe(exitUnlessInterrupted);596 ).pipe(exitUnlessInterrupted);
562 const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt;597 const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt;
598 if (Exit.isSuccess(connectedExit)) {
599 return {
600 _tag: "Interrupted",
601 established: true,
602 stable: connectedForMs >= BACKOFF_RESET_AFTER_MS,
603 resetRetry: connectedExit.value,
604 } satisfies AttemptOutcome;
605 }
563 return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS);606 return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS);
564 }, Effect.ensuring(clearLease));607 }, Effect.ensuring(clearLease));
565608
566 const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) {609 const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) {
567 return yield* Effect.raceFirst(610 return yield* Effect.raceFirst(
568 Effect.sleep(delayMs),611 Effect.sleep(delayMs).pipe(Effect.as(false)),
569 Effect.gen(function* () {612 Effect.gen(function* () {
570 for (;;) {613 for (;;) {
571 const next = yield* Queue.take(signals);614 const next = yield* Queue.take(signals);
572 switch (next._tag) {615 switch (next._tag) {
616 case "Wakeup":
617 return ConnectionWakeups.isApplicationActiveWakeup(next.reason);
573 case "ConnectRequested":618 case "ConnectRequested":
574 case "DisconnectRequested":619 case "DisconnectRequested":
575 case "RetryRequested":620 case "RetryRequested":
576 case "NetworkChanged":621 case "NetworkChanged":
577 case "Wakeup":622 return false;
578 return;
579 }623 }
580 }624 }
581 }),625 }),
582 );626 );
583 });627 });
584628
585 const waitForSignal = Queue.take(signals);629 const waitForSignal = Queue.take(signals).pipe(
630 Effect.map(
631 (next) => next._tag === "Wakeup" && ConnectionWakeups.isApplicationActiveWakeup(next.reason),
632 ),
633 );
586634
587 const run = Effect.fnUntraced(function* () {635 const run = Effect.fnUntraced(function* () {
588 let failureCount = 0;636 let failureCount = 0;
589 let generation = 0;637 let generation = 0;
590 let latestFailure: ConnectionAttemptError | null = null;638 let latestFailure: ConnectionAttemptError | null = null;
591 let pendingRetry = Option.none<PendingRetryTrace>();639 let pendingRetry = Option.none<PendingRetryTrace>();
640 const resetRetryLadder = () => {
641 failureCount = 0;
642 pendingRetry = Option.none();
643 };
592644
593 for (;;) {645 for (;;) {
594 const currentIntent = yield* Ref.get(intent);646 const currentIntent = yield* Ref.get(intent);
595 if (!currentIntent.desired) {647 if (!currentIntent.desired) {
596 failureCount = 0;648 resetRetryLadder();
597 latestFailure = null;649 latestFailure = null;
598 pendingRetry = Option.none();
599 yield* clearLease;650 yield* clearLease;
600 yield* setState(availableState(currentIntent, generation));651 yield* setState(availableState(currentIntent, generation));
601 yield* waitForSignal;652 yield* waitForSignal;
...@@ -604,7 +655,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -604,7 +655,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
604 if (currentIntent.network === "offline") {655 if (currentIntent.network === "offline") {
605 yield* clearLease;656 yield* clearLease;
606 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));657 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));
607 yield* waitForSignal;658 const applicationActivated = yield* waitForSignal;
659 if (applicationActivated) {
660 resetRetryLadder();
661 }
608 continue;662 continue;
609 }663 }
610664
...@@ -616,12 +670,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -616,12 +670,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
616 if (outcome.established) {670 if (outcome.established) {
617 generation = nextGeneration;671 generation = nextGeneration;
618 if (outcome.stable) {672 if (outcome.stable) {
619 failureCount = 0;673 resetRetryLadder();
620 latestFailure = null;674 latestFailure = null;
621 pendingRetry = Option.none();
622 }675 }
623 }676 }
624 if (outcome._tag === "Interrupted") {677 if (outcome._tag === "Interrupted") {
678 if (outcome.resetRetry) {
679 resetRetryLadder();
680 }
625 continue;681 continue;
626 }682 }
627683
...@@ -640,7 +696,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -640,7 +696,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
640 lastFailure: error,696 lastFailure: error,
641 retryAt: null,697 retryAt: null,
642 });698 });
643 yield* waitForSignal;699 const applicationActivated = yield* waitForSignal;
700 if (applicationActivated) {
701 resetRetryLadder();
702 }
644 continue;703 continue;
645 }704 }
646705
...@@ -663,7 +722,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -663,7 +722,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
663 lastFailure: error,722 lastFailure: error,
664 retryAt: (yield* Clock.currentTimeMillis) + delayMs,723 retryAt: (yield* Clock.currentTimeMillis) + delayMs,
665 });724 });
666 yield* waitForRetrySignal(delayMs);725 const applicationActivated = yield* waitForRetrySignal(delayMs);
726 if (applicationActivated) {
727 resetRetryLadder();
728 }
667 }729 }
668 });730 });
669731
packages/client-runtime/src/connection/wakeups.ts+17-1
...@@ -2,7 +2,23 @@ import * as Context from "effect/Context";...@@ -2,7 +2,23 @@ import * as Context from "effect/Context";
2import * as Layer from "effect/Layer";2import * as Layer from "effect/Layer";
3import type * as Stream from "effect/Stream";3import type * as Stream from "effect/Stream";
44
5export type ConnectionWakeup = "application-active" | "credentials-changed";5export type ConnectionWakeup =
6 | "application-active"
7 | "application-active-probe"
8 | "application-active-reconnect"
9 | "credentials-changed";
10
11export function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean {
12 return (
13 reason === "application-active" ||
14 reason === "application-active-probe" ||
15 reason === "application-active-reconnect"
16 );
17}
18
19export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean {
20 return reason === "application-active" || reason === "application-active-probe";
21}
622
7export class ConnectionWakeups extends Context.Service<23export class ConnectionWakeups extends Context.Service<
8 ConnectionWakeups,24 ConnectionWakeups,
packages/client-runtime/src/state/shell-sync.test.ts+15
...@@ -329,6 +329,21 @@ describe("environment shell synchronization", () => {...@@ -329,6 +329,21 @@ describe("environment shell synchronization", () => {
329329
330 expect(yield* Ref.get(loaderCalls)).toBe(2);330 expect(yield* Ref.get(loaderCalls)).toBe(2);
331 expect(yield* Ref.get(subscriptionCount)).toBe(2);331 expect(yield* Ref.get(subscriptionCount)).toBe(2);
332
333 yield* Queue.offer(wakeups, "application-active-probe");
334 for (let attempt = 0; attempt < 100; attempt += 1) {
335 if ((yield* Ref.get(subscriptionCount)) >= 3) break;
336 yield* Effect.yieldNow;
337 }
338 expect(yield* Ref.get(loaderCalls)).toBe(3);
339 expect(yield* Ref.get(subscriptionCount)).toBe(3);
340
341 yield* Queue.offer(wakeups, "application-active-reconnect");
342 for (let attempt = 0; attempt < 10; attempt += 1) {
343 yield* Effect.yieldNow;
344 }
345 expect(yield* Ref.get(loaderCalls)).toBe(3);
346 expect(yield* Ref.get(subscriptionCount)).toBe(3);
332 }),347 }),
333 );348 );
334});349});
packages/client-runtime/src/state/shell.ts+1-1
...@@ -172,7 +172,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -172,7 +172,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
172 const foregroundResubscriptions = Option.match(wakeups, {172 const foregroundResubscriptions = Option.match(wakeups, {
173 onNone: () => Stream.never,173 onNone: () => Stream.never,
174 onSome: (service) =>174 onSome: (service) =>
175 service.changes.pipe(Stream.filter((reason) => reason === "application-active")),175 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
176 });176 });
177177
178 yield* setSynchronizing;178 yield* setSynchronizing;
packages/client-runtime/src/state/threads-sync.test.ts+13
...@@ -696,6 +696,19 @@ describe("EnvironmentThreads", () => {...@@ -696,6 +696,19 @@ describe("EnvironmentThreads", () => {
696 (value) => value.status === "live" && Option.isSome(value.data),696 (value) => value.status === "live" && Option.isSome(value.data),
697 );697 );
698 expect(Option.getOrThrow(live.data).title).toBe("Latest title");698 expect(Option.getOrThrow(live.data).title).toBe("Latest title");
699
700 yield* Queue.offer(harness.wakeups, "application-active-probe");
701 for (let attempt = 0; attempt < 100; attempt += 1) {
702 if ((yield* Ref.get(harness.subscriptionCount)) >= 3) break;
703 yield* Effect.yieldNow;
704 }
705 expect(yield* Ref.get(harness.subscriptionCount)).toBe(3);
706
707 yield* Queue.offer(harness.wakeups, "application-active-reconnect");
708 for (let attempt = 0; attempt < 10; attempt += 1) {
709 yield* Effect.yieldNow;
710 }
711 expect(yield* Ref.get(harness.subscriptionCount)).toBe(3);
699 }),712 }),
700 );713 );
701});714});
packages/client-runtime/src/state/threads.ts+1-1
...@@ -236,7 +236,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -236,7 +236,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
236 const foregroundResubscriptions = Option.match(wakeups, {236 const foregroundResubscriptions = Option.match(wakeups, {
237 onNone: () => Stream.never,237 onNone: () => Stream.never,
238 onSome: (service) =>238 onSome: (service) =>
239 service.changes.pipe(Stream.filter((reason) => reason === "application-active")),239 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
240 });240 });
241241
242 yield* setSynchronizing;242 yield* setSynchronizing;