authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-02 21:42:36-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-02 21:42:36-07:00
log9333509c918083cbfb7e66759ddbcbfa71a3a452
tree8618c345a1304bf50baea5cb0751ad7cf16a4e2f
parentd668ffcd64d74daba16520cbbfb0fdb431983eac
signature Signed by PGP key B5690EEEBB952194

fix(client-runtime): reconnects back off with jitter and keep healthy sockets (#14897)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

9 files changed, 631 insertions(+), 148 deletions(-)

apps/server/src/telemetry/AnalyticsService.test.ts+83
...@@ -4,6 +4,10 @@ import { assert, it } from "@effect/vitest";...@@ -4,6 +4,10 @@ import { assert, it } from "@effect/vitest";
4import * as ConfigProvider from "effect/ConfigProvider";4import * as ConfigProvider from "effect/ConfigProvider";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8import * as TestClock from "effect/testing/TestClock";
9import * as HttpClient from "effect/unstable/http/HttpClient";
10import * as HttpClientError from "effect/unstable/http/HttpClientError";
7import * as HttpServer from "effect/unstable/http/HttpServer";11import * as HttpServer from "effect/unstable/http/HttpServer";
8import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";12import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";13import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";
...@@ -46,7 +50,86 @@ interface RecordedBatchBody {...@@ -46,7 +50,86 @@ interface RecordedBatchBody {
46 }>;50 }>;
47}51}
4852
53const SentBatch = Schema.fromJsonString(
54 Schema.Struct({
55 batch: Schema.Array(Schema.Struct({ uuid: Schema.String })),
56 }),
57);
58
59/**
60 * HTTP client that reads each batch, then fails as if the connection dropped
61 * before the response arrived. PostHog stores these batches, so the server
62 * must not send them forever.
63 */
64const acceptThenFailClient = (batches: Array<ReadonlyArray<{ readonly uuid: string }>>) =>
65 Layer.succeed(
66 HttpClient.HttpClient,
67 HttpClient.make((request) =>
68 Effect.gen(function* () {
69 if (request.body._tag === "Uint8Array") {
70 const body = yield* Schema.decodeEffect(SentBatch)(
71 new TextDecoder().decode(request.body.body),
72 ).pipe(Effect.orDie);
73 batches.push(body.batch);
74 }
75 return yield* new HttpClientError.HttpClientError({
76 reason: new HttpClientError.TransportError({ request, cause: "connection reset" }),
77 });
78 }),
79 ),
80 );
81
82it("retryDelayMs doubles from 2s and stays under the 5 minute cap", () => {
83 assert.equal(AnalyticsService.retryDelayMs(1, 0), 1_000);
84 assert.equal(AnalyticsService.retryDelayMs(2, 0.999_999), 4_000);
85 assert.equal(AnalyticsService.retryDelayMs(30, 0), 150_000);
86 assert.equal(AnalyticsService.retryDelayMs(30, 0.999_999), 300_000);
87});
88
49it.layer(NodeServices.layer)("AnalyticsService test", (it) => {89it.layer(NodeServices.layer)("AnalyticsService test", (it) => {
90 it.effect("a batch that keeps failing is retried with backoff, then dropped", () =>
91 Effect.gen(function* () {
92 const batches: Array<ReadonlyArray<{ readonly uuid: string }>> = [];
93 const runtimeLayer = AnalyticsService.layer.pipe(
94 Layer.provideMerge(
95 ServerConfig.ServerConfig.layerTest(process.cwd(), { prefix: "t3-telemetry-retry-" }),
96 ),
97 Layer.provide(
98 ConfigProvider.layer(
99 ConfigProvider.fromUnknown({
100 T3CODE_TELEMETRY_ENABLED: true,
101 T3CODE_POSTHOG_KEY: "phc_test_key",
102 T3CODE_POSTHOG_HOST: "http://localhost",
103 }),
104 ),
105 ),
106 Layer.provide(
107 Layer.mergeAll(
108 Layer.succeed(HostProcessPlatform, "win32"),
109 Layer.succeed(HostProcessArchitecture, "x64"),
110 acceptThenFailClient(batches),
111 ),
112 ),
113 );
114
115 yield* Effect.gen(function* () {
116 const analytics = yield* AnalyticsService.AnalyticsService;
117 for (let index = 0; index < 20; index += 1) {
118 yield* analytics.record("test.retry", { index });
119 }
120 // Before the fix this loop sent the batch about once a second.
121 for (let second = 0; second < 600; second += 1) {
122 yield* TestClock.adjust("1 second");
123 }
124 }).pipe(Effect.provide(runtimeLayer));
125
126 assert.equal(batches.length, 5);
127 const uuids = batches.map((batch) => batch.map((event) => event.uuid).join(","));
128 assert.equal(new Set(uuids).size, 1, "every retry carries the same uuids");
129 assert.equal(new Set(batches[0]?.map((event) => event.uuid)).size, 20);
130 }),
131 );
132
50 it.effect("flush drains all buffered events across multiple batches", () =>133 it.effect("flush drains all buffered events across multiple batches", () =>
51 Effect.gen(function* () {134 Effect.gen(function* () {
52 const capturedRequests: Array<RecordedBatchRequest> = [];135 const capturedRequests: Array<RecordedBatchRequest> = [];
apps/server/src/telemetry/AnalyticsService.ts+103-21
...@@ -2,19 +2,27 @@...@@ -2,19 +2,27 @@
2 * Anonymous PostHog telemetry service.2 * Anonymous PostHog telemetry service.
3 *3 *
4 * Persists an installation-scoped anonymous identifier, buffers events in4 * Persists an installation-scoped anonymous identifier, buffers events in
5 * memory, and flushes batches over Effect's HTTP client.5 * memory, and flushes batches over Effect's HTTP client. A failed batch is
6 * retried with backoff and dropped after a few tries. Each event carries a
7 * uuid, so PostHog can tell a retried copy from a new event.
6 *8 *
7 * @module AnalyticsService9 * @module AnalyticsService
8 */10 */
9import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";11import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";
10import type { ClientOs } from "@t3tools/contracts";12import type { ClientOs } from "@t3tools/contracts";
13import * as Clock from "effect/Clock";
11import * as Config from "effect/Config";14import * as Config from "effect/Config";
12import * as Context from "effect/Context";15import * as Context from "effect/Context";
16import * as Crypto from "effect/Crypto";
13import * as DateTime from "effect/DateTime";17import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";18import * as Effect from "effect/Effect";
19import * as Exit from "effect/Exit";
15import * as Layer from "effect/Layer";20import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";21import * as Option from "effect/Option";
22import * as Random from "effect/Random";
17import * as Ref from "effect/Ref";23import * as Ref from "effect/Ref";
24import * as Result from "effect/Result";
25import * as Semaphore from "effect/Semaphore";
18import * as HttpClient from "effect/unstable/http/HttpClient";26import * as HttpClient from "effect/unstable/http/HttpClient";
19import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";27import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
20import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";28import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";
...@@ -24,11 +32,40 @@ import * as ServerConfig from "../config.ts";...@@ -24,11 +32,40 @@ import * as ServerConfig from "../config.ts";
24import { getTelemetryIdentifier } from "./Identify.ts";32import { getTelemetryIdentifier } from "./Identify.ts";
2533
26interface BufferedAnalyticsEvent {34interface BufferedAnalyticsEvent {
35 readonly uuid: string;
27 readonly event: string;36 readonly event: string;
28 readonly properties?: Readonly<Record<string, unknown>>;37 readonly properties?: Readonly<Record<string, unknown>>;
29 readonly capturedAt: string;38 readonly capturedAt: string;
30}39}
3140
41interface DeliveryState {
42 /** Batch that failed last. It is sent again before newer events. */
43 readonly failedBatch: ReadonlyArray<BufferedAnalyticsEvent>;
44 /** Failed sends of `failedBatch`. */
45 readonly batchAttempts: number;
46 /** Failed sends since the last success. Sets the backoff delay. */
47 readonly failures: number;
48 /** The background flush does not send before this time (epoch ms). */
49 readonly retryAt: number;
50}
51
52const FLUSH_INTERVAL_MS = 1_000;
53// A hung send would hold the flush lock, and with it the shutdown flush.
54const SEND_TIMEOUT = "10 seconds";
55const MAX_BATCH_ATTEMPTS = 5;
56const RETRY_BASE_DELAY_MS = 2_000;
57const RETRY_MAX_DELAY_MS = 300_000;
58
59/**
60 * Delay before the next send after `failures` consecutive failed sends. The
61 * ceiling doubles from 2s up to 5 minutes, and the delay is a random point in
62 * its upper half. `random` is in [0, 1).
63 */
64export function retryDelayMs(failures: number, random: number): number {
65 const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failures - 1));
66 return Math.round(ceiling / 2 + (ceiling / 2) * random);
67}
68
32const TelemetryEnvConfig = Config.all({69const TelemetryEnvConfig = Config.all({
33 posthogKey: Config.String("T3CODE_POSTHOG_KEY").pipe(70 posthogKey: Config.String("T3CODE_POSTHOG_KEY").pipe(
34 Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),71 Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),
...@@ -88,17 +125,31 @@ export const make = Effect.gen(function* () {...@@ -88,17 +125,31 @@ export const make = Effect.gen(function* () {
88 const httpClient = yield* HttpClient.HttpClient;125 const httpClient = yield* HttpClient.HttpClient;
89 const serverConfig = yield* ServerConfig.ServerConfig;126 const serverConfig = yield* ServerConfig.ServerConfig;
90 const identifier = yield* getTelemetryIdentifier;127 const identifier = yield* getTelemetryIdentifier;
128 const crypto = yield* Crypto.Crypto;
91 const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);129 const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
130 const deliveryRef = yield* Ref.make<DeliveryState>({
131 failedBatch: [],
132 batchAttempts: 0,
133 failures: 0,
134 retryAt: 0,
135 });
136 // The background flush and the shutdown flush must not send the same batch at once.
137 const flushLock = yield* Semaphore.make(1);
92 const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";138 const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";
93 const hostPlatform = yield* HostProcessPlatform;139 const hostPlatform = yield* HostProcessPlatform;
94 const hostArchitecture = yield* HostProcessArchitecture;140 const hostArchitecture = yield* HostProcessArchitecture;
95141
96 const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) =>142 const enqueueBufferedEvent = (
143 uuid: string,
144 event: string,
145 properties?: Readonly<Record<string, unknown>>,
146 ) =>
97 Effect.flatMap(DateTime.now, (now) =>147 Effect.flatMap(DateTime.now, (now) =>
98 Ref.modify(bufferRef, (current) => {148 Ref.modify(bufferRef, (current) => {
99 const appended = [149 const appended = [
100 ...current,150 ...current,
101 {151 {
152 uuid,
102 event,153 event,
103 ...(properties ? { properties } : {}),154 ...(properties ? { properties } : {}),
104 capturedAt: DateTime.formatIso(now),155 capturedAt: DateTime.formatIso(now),
...@@ -128,6 +179,7 @@ export const make = Effect.gen(function* () {...@@ -128,6 +179,7 @@ export const make = Effect.gen(function* () {
128 const payload = {179 const payload = {
129 api_key: telemetryConfig.posthogKey,180 api_key: telemetryConfig.posthogKey,
130 batch: events.map((event) => ({181 batch: events.map((event) => ({
182 uuid: event.uuid,
131 event: event.event,183 event: event.event,
132 distinct_id: identifier,184 distinct_id: identifier,
133 properties: {185 properties: {
...@@ -152,39 +204,69 @@ export const make = Effect.gen(function* () {...@@ -152,39 +204,69 @@ export const make = Effect.gen(function* () {
152 HttpClientRequest.bodyJson(payload),204 HttpClientRequest.bodyJson(payload),
153 Effect.flatMap(httpClient.execute),205 Effect.flatMap(httpClient.execute),
154 Effect.flatMap(HttpClientResponse.filterStatusOk),206 Effect.flatMap(HttpClientResponse.filterStatusOk),
207 Effect.timeout(SEND_TIMEOUT),
155 );208 );
156 });209 });
157210
211 const takeBatch = Ref.modify(bufferRef, (current) => {
212 const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
213 return [nextBatch, current.slice(nextBatch.length)] as const;
214 });
215
216 // Sends batches until the buffer is empty or a send fails. A failed batch is
217 // kept for the next flush, and dropped after MAX_BATCH_ATTEMPTS failed sends.
158 const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () {218 const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () {
159 while (true) {219 while (true) {
160 const batch = yield* Ref.modify(bufferRef, (current) => {220 const delivery = yield* Ref.get(deliveryRef);
161 if (current.length === 0) {221 const batch = delivery.failedBatch.length > 0 ? delivery.failedBatch : yield* takeBatch;
162 return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const;
163 }
164 const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
165 const remaining = current.slice(nextBatch.length);
166 return [nextBatch, remaining] as const;
167 });
168
169 if (batch.length === 0) {222 if (batch.length === 0) {
170 return;223 return;
171 }224 }
172225
173 yield* sendBatch(batch).pipe(226 const sent = yield* Effect.result(sendBatch(batch));
174 Effect.catch((error) =>227 if (Result.isSuccess(sent)) {
175 Ref.update(bufferRef, (current) => [...batch, ...current]).pipe(228 yield* Ref.set(deliveryRef, { failedBatch: [], batchAttempts: 0, failures: 0, retryAt: 0 });
176 Effect.flatMap(() => Effect.fail(error)),229 continue;
177 ),230 }
178 ),231
179 );232 const failures = delivery.failures + 1;
233 const batchAttempts = delivery.batchAttempts + 1;
234 const retryAt = (yield* Clock.currentTimeMillis) + retryDelayMs(failures, yield* Random.next);
235 if (batchAttempts < MAX_BATCH_ATTEMPTS) {
236 yield* Ref.set(deliveryRef, { failedBatch: batch, batchAttempts, failures, retryAt });
237 yield* Effect.logDebug("Failed to send telemetry batch; will retry", {
238 attempt: batchAttempts,
239 cause: sent.failure,
240 });
241 return;
242 }
243 yield* Ref.set(deliveryRef, { failedBatch: [], batchAttempts: 0, failures, retryAt });
244 yield* Effect.logWarning("Dropped telemetry batch after repeated send failures", {
245 events: batch.length,
246 attempts: batchAttempts,
247 cause: sent.failure,
248 });
249 return;
180 }250 }
181 }).pipe(Effect.catch((cause) => Effect.logError("Failed to flush telemetry", { cause })));251 }).pipe(flushLock.withPermit);
252
253 const flushWhenDue = Effect.gen(function* () {
254 const { retryAt } = yield* Ref.get(deliveryRef);
255 if ((yield* Clock.currentTimeMillis) >= retryAt) {
256 yield* flush;
257 }
258 });
182259
183 const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")(260 const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")(
184 function* (event, properties) {261 function* (event, properties) {
185 if (!telemetryConfig.enabled || !identifier) return;262 if (!telemetryConfig.enabled || !identifier) return;
186263
187 const enqueueResult = yield* enqueueBufferedEvent(event, properties);264 // Telemetry is best effort: an event without a uuid is not sent. The
265 // Node implementation throws (a defect) rather than failing, so catch both.
266 const uuid = yield* Effect.exit(crypto.randomUUIDv7);
267 if (Exit.isFailure(uuid)) return;
268
269 const enqueueResult = yield* enqueueBufferedEvent(uuid.value, event, properties);
188 if (enqueueResult.dropped) {270 if (enqueueResult.dropped) {
189 yield* Effect.logDebug("analytics buffer full; dropping oldest event", {271 yield* Effect.logDebug("analytics buffer full; dropping oldest event", {
190 size: enqueueResult.size,272 size: enqueueResult.size,
...@@ -194,7 +276,7 @@ export const make = Effect.gen(function* () {...@@ -194,7 +276,7 @@ export const make = Effect.gen(function* () {
194 },276 },
195 );277 );
196278
197 yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), {279 yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.flatMap(() => flushWhenDue)), {
198 disableYield: true,280 disableYield: true,
199 }).pipe(Effect.forkScoped);281 }).pipe(Effect.forkScoped);
200282
docs/internals/connection-runtime.md+13-9
...@@ -10,16 +10,20 @@ several views need the same environment....@@ -10,16 +10,20 @@ several views need the same environment.
1010
11The [supervisor](../../packages/client-runtime/src/connection/supervisor.ts) owns11The [supervisor](../../packages/client-runtime/src/connection/supervisor.ts) owns
12transport retry policy; resolving an endpoint and opening an RPC session are single12transport retry policy; resolving an endpoint and opening an RPC session are single
13attempts. Transient failures retry with capped backoff. Offline states and13attempts. Transient failures retry with jittered exponential backoff, capped at
14authentication failures wait for a wakeup instead of spending attempts on14five minutes, that resets only after a connection stays up. Without jitter, every
15unchanged conditions.15client of a restarted server reconnects in the same second; with a short cap, a
16client that can never connect retries all day. Offline states and authentication
17failures wait for a wakeup instead of spending attempts on unchanged conditions.
1618
17Foregrounding needs different treatment depending on the connection's state.19Foregrounding, an explicit retry, and an offline report probe the established
18It wakes a retry immediately, leaves an ordinary in-flight attempt alone, and20session, and only a failed probe reconnects. Offline reports are often wrong, for
19probes an established session before replacing it. A long mobile background21example for a loopback server. A long mobile background suspension is the one
20suspension forces replacement because the OS can kill a socket without reporting22exception: it replaces the session at once, because the OS can kill a socket
21closure. Treating every foreground event as a reconnect delays healthy attempts;23without reporting closure, and a probe would hold a dead socket in "Resuming"
22treating every resume as harmless leaves suspended sockets stuck.24until it times out. That fresh attempt runs even while the network reports
25offline. Foregrounding also wakes a pending retry immediately and
26leaves an ordinary in-flight attempt alone.
2327
24The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes28The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes
25connections by environment. An involuntary disconnect retains the registration29connections by environment. An involuntary disconnect retains the registration
docs/internals/product-analytics.md+7
...@@ -41,6 +41,13 @@ Unknown counts stay absent. Partial usage contains valid observed counts but...@@ -41,6 +41,13 @@ Unknown counts stay absent. Partial usage contains valid observed counts but
41cannot establish a whole-turn total. Keep these distinctions when changing token41cannot establish a whole-turn total. Keep these distinctions when changing token
42normalization or building reports.42normalization or building reports.
4343
44## Delivery
45
46A send can fail after PostHog has stored the batch, so every retry is a copy.
47[Delivery](../../apps/server/src/telemetry/AnalyticsService.ts) gives each event a
48uuid when it is recorded, backs off after a failed send, and drops a batch after a
49few tries. Without these limits, one stuck batch was sent every second for days.
50
44## Collection boundary51## Collection boundary
4552
46Keep analytics payloads to product metadata and normalized measurements. Do not53Keep analytics payloads to product metadata and normalized measurements. Do not
packages/client-runtime/src/connection/registry.test.ts+40
...@@ -1443,6 +1443,46 @@ describe("EnvironmentRegistry", () => {...@@ -1443,6 +1443,46 @@ describe("EnvironmentRegistry", () => {
1443 }),1443 }),
1444 );1444 );
14451445
1446 it.effect("keeps one session per environment across concurrent registrations and retries", () =>
1447 Effect.gen(function* () {
1448 const harness = yield* makeHarness([]);
1449
1450 yield* Effect.gen(function* () {
1451 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
1452 const registration = new PrimaryConnectionRegistration({ target: TARGET });
1453 yield* Effect.all(
1454 Array.from({ length: 5 }, () => registry.registerPlatform(registration)),
1455 { concurrency: "unbounded", discard: true },
1456 );
1457 yield* awaitConnectionState(
1458 registry,
1459 TARGET.environmentId,
1460 (state) => state.phase === "connected",
1461 );
1462
1463 // Platform polls and explicit retries reach a healthy connection at once.
1464 yield* Effect.all(
1465 [
1466 ...Array.from({ length: 5 }, () => registry.registerPlatform(registration)),
1467 ...Array.from({ length: 5 }, () => registry.retryNow(TARGET.environmentId)),
1468 registry.reconcilePlatform([registration]),
1469 ],
1470 { concurrency: "unbounded", discard: true },
1471 );
1472 for (let attempt = 0; attempt < 100; attempt += 1) {
1473 yield* Effect.yieldNow;
1474 }
1475
1476 expect(yield* Ref.get(harness.sessions)).toHaveLength(1);
1477 expect(yield* Ref.get(harness.releasedSessions)).toBe(0);
1478 expect(yield* registry.state(TARGET.environmentId)).toMatchObject({
1479 phase: "connected",
1480 generation: 1,
1481 });
1482 }).pipe(Effect.provide(harness.layer), Effect.scoped);
1483 }),
1484 );
1485
1446 it.effect("retains a healthy runtime when the platform repeats an identical registration", () =>1486 it.effect("retains a healthy runtime when the platform repeats an identical registration", () =>
1447 Effect.gen(function* () {1487 Effect.gen(function* () {
1448 const harness = yield* makeHarness([]);1488 const harness = yield* makeHarness([]);
packages/client-runtime/src/connection/supervisor.test.ts+232-13
...@@ -5,6 +5,7 @@ import * as Deferred from "effect/Deferred";...@@ -5,6 +5,7 @@ import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";7import * as Option from "effect/Option";
8import * as Random from "effect/Random";
8import * as Ref from "effect/Ref";9import * as Ref from "effect/Ref";
9import * as Stream from "effect/Stream";10import * as Stream from "effect/Stream";
10import * as SubscriptionRef from "effect/SubscriptionRef";11import * as SubscriptionRef from "effect/SubscriptionRef";
...@@ -186,6 +187,11 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:...@@ -186,6 +187,11 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
186 });187 });
187188
188 const dependencies = Layer.mergeAll(189 const dependencies = Layer.mergeAll(
190 // Jitter at its maximum, so each retry waits exactly its ceiling: 2s, 4s, 8s...
191 Layer.succeed(Random.Random, {
192 nextDoubleUnsafe: () => 1 - Number.EPSILON,
193 nextIntUnsafe: () => 0,
194 }),
189 Layer.succeed(Connectivity.Connectivity, connectivity),195 Layer.succeed(Connectivity.Connectivity, connectivity),
190 Layer.succeed(196 Layer.succeed(
191 ConnectionWakeups.ConnectionWakeups,197 ConnectionWakeups.ConnectionWakeups,
...@@ -225,6 +231,17 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:...@@ -225,6 +231,17 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
225 };231 };
226});232});
227233
234describe("retryDelayMs", () => {
235 it("doubles from 2 seconds to a 5 minute cap, jittered within the upper half of each step", () => {
236 const ceilings = [2_000, 4_000, 8_000, 16_000, 32_000, 64_000, 128_000, 256_000, 300_000];
237 for (const [failureCount, ceiling] of [...ceilings, 300_000].entries()) {
238 expect(EnvironmentSupervisor.retryDelayMs(failureCount, 0)).toBe(ceiling / 2);
239 expect(EnvironmentSupervisor.retryDelayMs(failureCount, 0.5)).toBe((ceiling * 3) / 4);
240 expect(EnvironmentSupervisor.retryDelayMs(failureCount, 1 - Number.EPSILON)).toBe(ceiling);
241 }
242 });
243});
244
228describe("EnvironmentSupervisor", () => {245describe("EnvironmentSupervisor", () => {
229 it.effect("exports each relay setup as a standalone linked trace that ends at readiness", () =>246 it.effect("exports each relay setup as a standalone linked trace that ends at readiness", () =>
230 Effect.gen(function* () {247 Effect.gen(function* () {
...@@ -353,7 +370,7 @@ describe("EnvironmentSupervisor", () => {...@@ -353,7 +370,7 @@ describe("EnvironmentSupervisor", () => {
353 }),370 }),
354 );371 );
355372
356 it.effect("retries forever with exponential backoff capped at sixteen seconds", () =>373 it.effect("retries forever with exponential backoff capped at five minutes", () =>
357 Effect.gen(function* () {374 Effect.gen(function* () {
358 const harness = yield* makeHarness({375 const harness = yield* makeHarness({
359 prepare: () => Effect.fail(transient()),376 prepare: () => Effect.fail(transient()),
...@@ -368,15 +385,20 @@ describe("EnvironmentSupervisor", () => {...@@ -368,15 +385,20 @@ describe("EnvironmentSupervisor", () => {
368 );385 );
369 expect(yield* Ref.get(harness.prepareCount)).toBe(1);386 expect(yield* Ref.get(harness.prepareCount)).toBe(1);
370387
371 for (const [index, delay] of [3_000, 4_000, 8_000, 16_000, 16_000, 16_000].entries()) {388 const delays = [
372 yield* TestClock.adjust(delay);389 2_000, 4_000, 8_000, 16_000, 32_000, 64_000, 128_000, 256_000, 300_000, 300_000,
390 ];
391 for (const [index, delay] of delays.entries()) {
392 yield* TestClock.adjust(delay - 1);
393 expect(yield* Ref.get(harness.prepareCount)).toBe(index + 1);
394 yield* TestClock.adjust(1);
373 yield* eventuallyState(395 yield* eventuallyState(
374 supervisor.state,396 supervisor.state,
375 (state) => state.phase === "backoff" && state.attempt === index + 2,397 (state) => state.phase === "backoff" && state.attempt === index + 2,
376 );398 );
377 }399 }
378400
379 expect(yield* Ref.get(harness.prepareCount)).toBe(7);401 expect(yield* Ref.get(harness.prepareCount)).toBe(delays.length + 1);
380 }).pipe(Effect.provide(TestClock.layer())),402 }).pipe(Effect.provide(TestClock.layer())),
381 );403 );
382404
...@@ -578,7 +600,7 @@ describe("EnvironmentSupervisor", () => {...@@ -578,7 +600,7 @@ describe("EnvironmentSupervisor", () => {
578 );600 );
579 expect(yield* Ref.get(harness.prepareCount)).toBe(3);601 expect(yield* Ref.get(harness.prepareCount)).toBe(3);
580602
581 yield* TestClock.adjust("2999 millis");603 yield* TestClock.adjust("1999 millis");
582 expect(yield* Ref.get(harness.prepareCount)).toBe(3);604 expect(yield* Ref.get(harness.prepareCount)).toBe(3);
583 yield* TestClock.adjust("1 milli");605 yield* TestClock.adjust("1 milli");
584 yield* eventuallyState(606 yield* eventuallyState(
...@@ -641,9 +663,12 @@ describe("EnvironmentSupervisor", () => {...@@ -641,9 +663,12 @@ describe("EnvironmentSupervisor", () => {
641 }).pipe(Effect.provide(TestClock.layer())),663 }).pipe(Effect.provide(TestClock.layer())),
642 );664 );
643665
644 it.effect("releases a live session while offline and starts a new generation when online", () =>666 it.effect("keeps a session that still answers when the network reports offline", () =>
645 Effect.gen(function* () {667 Effect.gen(function* () {
646 const harness = yield* makeHarness();668 const probeCount = yield* Ref.make(0);
669 const harness = yield* makeHarness({
670 probe: () => Ref.update(probeCount, (count) => count + 1),
671 });
647 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {672 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
648 initiallyDesired: true,673 initiallyDesired: true,
649 }).pipe(Effect.provide(harness.dependencies));674 }).pipe(Effect.provide(harness.dependencies));
...@@ -652,21 +677,160 @@ describe("EnvironmentSupervisor", () => {...@@ -652,21 +677,160 @@ describe("EnvironmentSupervisor", () => {
652 supervisor.state,677 supervisor.state,
653 (state) => state.phase === "connected" && state.generation === 1,678 (state) => state.phase === "connected" && state.generation === 1,
654 );679 );
680 // A loopback server, or a flap shorter than the probe, keeps working.
655 yield* harness.setNetworkStatus("offline");681 yield* harness.setNetworkStatus("offline");
656 yield* awaitState(supervisor.state, (state) => state.phase === "offline");682 for (let attempt = 0; attempt < 100; attempt += 1) {
683 if ((yield* Ref.get(probeCount)) > 0) break;
684 yield* Effect.yieldNow;
685 }
686 yield* harness.setNetworkStatus("online");
687 yield* Effect.yieldNow;
657688
658 expect(yield* Ref.get(harness.releaseCount)).toBe(1);689 expect(yield* Ref.get(probeCount)).toBe(1);
659 expect(Option.isNone(yield* SubscriptionRef.get(supervisor.session))).toBe(true);690 expect(yield* Ref.get(harness.sessionCount)).toBe(1);
691 expect(yield* Ref.get(harness.releaseCount)).toBe(0);
692 expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({
693 phase: "connected",
694 generation: 1,
695 });
696 }),
697 );
698
699 it.effect("replaces the session on a long resume while the network reports offline", () =>
700 Effect.gen(function* () {
701 const probeCount = yield* Ref.make(0);
702 const harness = yield* makeHarness({
703 probe: () => Ref.update(probeCount, (count) => count + 1),
704 });
705 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
706 initiallyDesired: true,
707 }).pipe(Effect.provide(harness.dependencies));
660708
661 yield* harness.setNetworkStatus("online");
662 yield* awaitState(709 yield* awaitState(
710 supervisor.state,
711 (state) => state.phase === "connected" && state.generation === 1,
712 );
713 // A wrong offline report: the probe answers, so the session stays.
714 yield* harness.setNetworkStatus("offline");
715 for (let attempt = 0; attempt < 100; attempt += 1) {
716 if ((yield* Ref.get(probeCount)) > 0) break;
717 yield* Effect.yieldNow;
718 }
719 expect(yield* Ref.get(harness.sessionCount)).toBe(1);
720
721 // The replacement connects although the network still reports offline.
722 yield* harness.wake("application-active-reconnect");
723 const replaced = yield* awaitState(
663 supervisor.state,724 supervisor.state,
664 (state) => state.phase === "connected" && state.generation === 2,725 (state) => state.phase === "connected" && state.generation === 2,
665 );726 );
727
728 expect(replaced.attempt).toBe(1);
729 expect(yield* Ref.get(probeCount)).toBe(1);
666 expect(yield* Ref.get(harness.sessionCount)).toBe(2);730 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
731 expect(yield* Ref.get(harness.releaseCount)).toBe(1);
732 }),
733 );
734
735 it.effect(
736 "releases a session that stops answering while offline and reconnects when online",
737 () =>
738 Effect.gen(function* () {
739 const harness = yield* makeHarness({
740 probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
741 });
742 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
743 initiallyDesired: true,
744 }).pipe(Effect.provide(harness.dependencies));
745
746 yield* awaitState(
747 supervisor.state,
748 (state) => state.phase === "connected" && state.generation === 1,
749 );
750 yield* harness.setNetworkStatus("offline");
751 yield* TestClock.adjust("3 seconds");
752 yield* awaitState(supervisor.state, (state) => state.phase === "offline");
753
754 expect(yield* Ref.get(harness.releaseCount)).toBe(1);
755 expect(Option.isNone(yield* SubscriptionRef.get(supervisor.session))).toBe(true);
756
757 yield* harness.setNetworkStatus("online");
758 yield* awaitState(
759 supervisor.state,
760 (state) => state.phase === "connected" && state.generation === 2,
761 );
762 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
763 }).pipe(Effect.provide(TestClock.layer())),
764 );
765
766 it.effect("probes instead of replacing a healthy session on an explicit retry", () =>
767 Effect.gen(function* () {
768 const probeCount = yield* Ref.make(0);
769 const harness = yield* makeHarness({
770 probe: () => Ref.update(probeCount, (count) => count + 1),
771 });
772 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
773 initiallyDesired: true,
774 }).pipe(Effect.provide(harness.dependencies));
775
776 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
777 yield* supervisor.retryNow;
778 for (let attempt = 0; attempt < 100; attempt += 1) {
779 if ((yield* Ref.get(probeCount)) > 0) break;
780 yield* Effect.yieldNow;
781 }
782
783 expect(yield* Ref.get(probeCount)).toBe(1);
784 expect(yield* Ref.get(harness.sessionCount)).toBe(1);
785 expect(yield* Ref.get(harness.releaseCount)).toBe(0);
667 }),786 }),
668 );787 );
669788
789 it.effect("keeps the backoff ladder after an explicit retry finds a healthy session", () =>
790 Effect.gen(function* () {
791 const probeCount = yield* Ref.make(0);
792 const harness = yield* makeHarness({
793 probe: () => Ref.update(probeCount, (count) => count + 1),
794 });
795 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
796 initiallyDesired: true,
797 }).pipe(Effect.provide(harness.dependencies));
798
799 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
800 yield* harness.closeLatestSession();
801 yield* awaitState(
802 supervisor.state,
803 (state) => state.phase === "backoff" && state.attempt === 1,
804 );
805 yield* TestClock.adjust("2 seconds");
806 yield* awaitState(
807 supervisor.state,
808 (state) => state.phase === "connected" && state.generation === 2,
809 );
810
811 yield* supervisor.retryNow;
812 for (let attempt = 0; attempt < 100; attempt += 1) {
813 if ((yield* Ref.get(probeCount)) > 0) break;
814 yield* Effect.yieldNow;
815 }
816 expect(yield* Ref.get(probeCount)).toBe(1);
817
818 // The flapping session keeps climbing the ladder: the answered retry does
819 // not reset it after this unrelated close.
820 yield* harness.closeLatestSession();
821 yield* awaitState(
822 supervisor.state,
823 (state) => state.phase === "backoff" && state.attempt === 2,
824 );
825 yield* TestClock.adjust("4 seconds");
826 const reconnected = yield* awaitState(
827 supervisor.state,
828 (state) => state.phase === "connected" && state.generation === 3,
829 );
830 expect(reconnected.attempt).toBe(3);
831 }).pipe(Effect.provide(TestClock.layer())),
832 );
833
670 it.effect("retries a blocked connection when platform credentials change", () =>834 it.effect("retries a blocked connection when platform credentials change", () =>
671 Effect.gen(function* () {835 Effect.gen(function* () {
672 const harness = yield* makeHarness({836 const harness = yield* makeHarness({
...@@ -844,7 +1008,7 @@ describe("EnvironmentSupervisor", () => {...@@ -844,7 +1008,7 @@ describe("EnvironmentSupervisor", () => {
844 supervisor.state,1008 supervisor.state,
845 (state) => state.phase === "backoff" && state.attempt === 1,1009 (state) => state.phase === "backoff" && state.attempt === 1,
846 );1010 );
847 yield* TestClock.adjust("3 seconds");1011 yield* TestClock.adjust("2 seconds");
848 yield* awaitState(1012 yield* awaitState(
849 supervisor.state,1013 supervisor.state,
850 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 2,1014 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 2,
...@@ -964,6 +1128,35 @@ describe("EnvironmentSupervisor", () => {...@@ -964,6 +1128,35 @@ describe("EnvironmentSupervisor", () => {
964 }),1128 }),
965 );1129 );
9661130
1131 it.effect("reconnects immediately when the session closes during a resume probe", () =>
1132 Effect.gen(function* () {
1133 const probeStarted = yield* Deferred.make<void>();
1134 const harness = yield* makeHarness({
1135 probe: (attempt) =>
1136 attempt === 1
1137 ? Deferred.succeed(probeStarted, undefined).pipe(Effect.andThen(Effect.never))
1138 : Effect.void,
1139 });
1140 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
1141 initiallyDesired: true,
1142 }).pipe(Effect.provide(harness.dependencies));
1143
1144 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
1145 yield* harness.wake("application-active-probe");
1146 yield* Deferred.await(probeStarted);
1147 // The OS reports the suspended socket's close before the probe answers.
1148 yield* harness.closeLatestSession();
1149
1150 // No TestClock advance: the unanswered probe skips the first backoff rung.
1151 const reconnected = yield* awaitState(
1152 supervisor.state,
1153 (state) => state.phase === "connected" && state.generation === 2,
1154 );
1155 expect(reconnected.attempt).toBe(1);
1156 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
1157 }).pipe(Effect.provide(TestClock.layer())),
1158 );
1159
967 it.effect("reconnects immediately when the foreground liveness probe fails", () =>1160 it.effect("reconnects immediately when the foreground liveness probe fails", () =>
968 Effect.gen(function* () {1161 Effect.gen(function* () {
969 const allowReconnect = yield* Deferred.make<void>();1162 const allowReconnect = yield* Deferred.make<void>();
...@@ -1020,7 +1213,7 @@ describe("EnvironmentSupervisor", () => {...@@ -1020,7 +1213,7 @@ describe("EnvironmentSupervisor", () => {
1020 supervisor.state,1213 supervisor.state,
1021 (state) => state.phase === "backoff" && state.attempt === 1,1214 (state) => state.phase === "backoff" && state.attempt === 1,
1022 );1215 );
1023 yield* TestClock.adjust("2999 millis");1216 yield* TestClock.adjust("1999 millis");
1024 expect(yield* Ref.get(harness.prepareCount)).toBe(2);1217 expect(yield* Ref.get(harness.prepareCount)).toBe(2);
1025 yield* TestClock.adjust("1 milli");1218 yield* TestClock.adjust("1 milli");
1026 yield* eventuallyState(1219 yield* eventuallyState(
...@@ -1055,6 +1248,32 @@ describe("EnvironmentSupervisor", () => {...@@ -1055,6 +1248,32 @@ describe("EnvironmentSupervisor", () => {
1055 }).pipe(Effect.provide(TestClock.layer())),1248 }).pipe(Effect.provide(TestClock.layer())),
1056 );1249 );
10571250
1251 it.effect("an explicit retry shortens a stalled desktop foreground probe", () =>
1252 Effect.gen(function* () {
1253 const harness = yield* makeHarness({
1254 probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
1255 });
1256 const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
1257 initiallyDesired: true,
1258 }).pipe(Effect.provide(harness.dependencies));
1259
1260 yield* awaitState(supervisor.state, (state) => state.phase === "connected");
1261 yield* harness.wake("application-active");
1262 yield* TestClock.adjust("5 seconds");
1263 yield* supervisor.retryNow;
1264 // The retry's 3 second limit applies, not the 10 seconds left of the 15.
1265 yield* TestClock.adjust("2999 millis");
1266 expect(yield* Ref.get(harness.sessionCount)).toBe(1);
1267 yield* TestClock.adjust("1 milli");
1268 yield* awaitState(
1269 supervisor.state,
1270 (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1,
1271 );
1272
1273 expect(yield* Ref.get(harness.sessionCount)).toBe(2);
1274 }).pipe(Effect.provide(TestClock.layer())),
1275 );
1276
1058 it.effect("quickly times out a stalled mobile foreground liveness probe", () =>1277 it.effect("quickly times out a stalled mobile foreground liveness probe", () =>
1059 Effect.gen(function* () {1278 Effect.gen(function* () {
1060 const harness = yield* makeHarness({1279 const harness = yield* makeHarness({
packages/client-runtime/src/connection/supervisor.ts+149-102
...@@ -2,11 +2,13 @@ import { withRelayClientTracing } from "@t3tools/shared/relayTracing";...@@ -2,11 +2,13 @@ import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
2import * as Cause from "effect/Cause";2import * as Cause from "effect/Cause";
3import * as Clock from "effect/Clock";3import * as Clock from "effect/Clock";
4import * as Context from "effect/Context";4import * as Context from "effect/Context";
5import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";7import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";8import * as Fiber from "effect/Fiber";
8import * as Option from "effect/Option";9import * as Option from "effect/Option";
9import * as Queue from "effect/Queue";10import * as Queue from "effect/Queue";
11import * as Random from "effect/Random";
10import * as Ref from "effect/Ref";12import * as Ref from "effect/Ref";
11import * as Scope from "effect/Scope";13import * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";14import * as Stream from "effect/Stream";
...@@ -29,10 +31,13 @@ import { safeErrorLogAttributes } from "../errors/safeLog.ts";...@@ -29,10 +31,13 @@ import { safeErrorLogAttributes } from "../errors/safeLog.ts";
29import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";31import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
30import * as ConnectionWakeups from "./wakeups.ts";32import * as ConnectionWakeups from "./wakeups.ts";
3133
32const RETRY_DELAYS_MS = [3_000, 4_000, 8_000, 16_000] as const;34const RETRY_BASE_DELAY_MS = 1_000;
35const RETRY_MAX_DELAY_MS = 300_000;
33const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";36const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";
34const CONNECTION_PROBE_TIMEOUT = "15 seconds";37const CONNECTION_PROBE_TIMEOUT = "15 seconds";
35const MOBILE_CONNECTION_PROBE_TIMEOUT = "3 seconds";38// Mobile resumes, explicit retries, and offline events want a fast answer:
39// the user is waiting, or the network may be gone.
40const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds";
36const BACKOFF_RESET_AFTER_MS = 30_000;41const BACKOFF_RESET_AFTER_MS = 30_000;
3742
38interface SupervisorIntent {43interface SupervisorIntent {
...@@ -101,8 +106,19 @@ export interface EnvironmentSupervisorOptions {...@@ -101,8 +106,19 @@ export interface EnvironmentSupervisorOptions {
101 readonly initiallyDesired?: boolean;106 readonly initiallyDesired?: boolean;
102}107}
103108
104function retryDelayMs(failureCount: number): number {109/**
105 return RETRY_DELAYS_MS[Math.min(failureCount, RETRY_DELAYS_MS.length - 1)] ?? 16_000;110 * Delay before the next attempt after `failureCount` consecutive failures
111 * (0 for the first retry). The ceiling doubles from 2s up to 5 minutes, and
112 * the delay is a random point in its upper half: never quicker than half the
113 * ceiling, and spread out so clients that lost the same server do not all
114 * reconnect in the same second. `random` is in [0, 1).
115 *
116 * The long cap only applies to a connection that keeps failing. Returning to
117 * the app, the network coming back, and an explicit retry all skip the wait.
118 */
119export function retryDelayMs(failureCount: number, random: number): number {
120 const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failureCount + 1));
121 return Math.round(ceiling / 2 + (ceiling / 2) * random);
106}122}
107123
108function annotateTarget(target: ConnectionTarget) {124function annotateTarget(target: ConnectionTarget) {
...@@ -235,10 +251,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -235,10 +251,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
235 const intent = yield* Ref.make(initialIntent);251 const intent = yield* Ref.make(initialIntent);
236 const signals = yield* Queue.unbounded<SupervisorSignal>();252 const signals = yield* Queue.unbounded<SupervisorSignal>();
237 const resetRetryState = yield* Ref.make(false);253 const resetRetryState = yield* Ref.make(false);
238 // Set when a foreground wake probe fails or times out: the user is actively254 // Set while a probe of the live session is running, and kept when it fails
239 // returning to the app on a dead transport, so the follow-up reconnect skips255 // or times out: something asked whether the connection still works and it
240 // the first backoff rung instead of sleeping.256 // closed or failed before answering, so the follow-up reconnect skips the
241 const wakeProbeFailed = yield* Ref.make(false);257 // first backoff rung instead of sleeping.
258 const probeUnanswered = yield* Ref.make(false);
242 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(259 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(
243 !initialIntent.desired260 !initialIntent.desired
244 ? availableState(initialIntent, 0)261 ? availableState(initialIntent, 0)
...@@ -392,96 +409,118 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -392,96 +409,118 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
392 }409 }
393 });410 });
394411
412 // Signals that end a connected lease whatever its health: "reset" ends it
413 // and restarts the retry ladder, "end" ends it, undefined keeps it.
414 const connectedLeaseEnd = Effect.fnUntraced(function* (next: SupervisorSignal) {
415 if (next._tag === "DisconnectRequested") {
416 return "end" as const;
417 }
418 if (next._tag !== "Wakeup") {
419 return undefined;
420 }
421 if (next.reason === "application-active-reconnect") {
422 // Mobile operating systems often kill a suspended socket without a close
423 // event. A probe would show a dead socket as "Resuming" until it times
424 // out, so a long background resume replaces the session at once.
425 return "reset" as const;
426 }
427 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {
428 yield* logManagedRelayAccountChange;
429 return "end" as const;
430 }
431 return undefined;
432 });
433
434 // How long a signal waits for the live session to answer a probe, or
435 // undefined when the signal does not question the connection.
436 const probeTimeoutFor = (next: SupervisorSignal): Duration.Input | undefined => {
437 switch (next._tag) {
438 case "RetryRequested":
439 return QUICK_CONNECTION_PROBE_TIMEOUT;
440 case "NetworkChanged":
441 return next.network === "offline" ? QUICK_CONNECTION_PROBE_TIMEOUT : undefined;
442 case "Wakeup":
443 if (next.reason === "application-active") {
444 return CONNECTION_PROBE_TIMEOUT;
445 }
446 return next.reason === "application-active-probe"
447 ? QUICK_CONNECTION_PROBE_TIMEOUT
448 : undefined;
449 case "ConnectRequested":
450 case "DisconnectRequested":
451 return undefined;
452 }
453 };
454
455 // Holds a connected lease until it must end, and returns whether to restart
456 // the retry ladder. Returning to the app, an explicit retry, and the network
457 // reporting offline all probe the live session instead of replacing it, so a
458 // healthy socket is not torn down (the offline report is often wrong, for
459 // example for a loopback server). Only a long mobile resume replaces the
460 // session without a probe. A failed probe fails this effect, and the
461 // supervisor reconnects.
395 const monitorConnectedLease = Effect.fnUntraced(function* (462 const monitorConnectedLease = Effect.fnUntraced(function* (
396 lease: ConnectionDriver.EnvironmentConnectionLease,463 lease: ConnectionDriver.EnvironmentConnectionLease,
397 ) {464 ) {
465 // A probe answers an explicit retry here, so the retry must not also reset
466 // the backoff of a later, unrelated failure.
467 const takeSignal = Queue.take(signals).pipe(
468 Effect.tap((next) =>
469 next._tag === "RetryRequested" ? Ref.set(resetRetryState, false) : Effect.void,
470 ),
471 );
398 for (;;) {472 for (;;) {
399 const next = yield* Queue.take(signals);473 const next = yield* takeSignal;
400 switch (next._tag) {474 const end = yield* connectedLeaseEnd(next);
401 case "DisconnectRequested":475 if (end !== undefined) {
402 case "RetryRequested":476 return end === "reset";
403 return false;477 }
404 case "NetworkChanged":478 const probeTimeout = probeTimeoutFor(next);
405 if (next.network === "offline") {479 if (probeTimeout === undefined) {
406 return false;480 continue;
407 }481 }
408 break;482 yield* Ref.set(probeUnanswered, true);
409 case "Wakeup":483 const probe = yield* Effect.forkChild(lease.session.probe);
410 if (next.reason === "credentials-changed" && target._tag === "RelayConnectionTarget") {484 // Monotonic nanoseconds, so a wall-clock correction cannot move the deadline.
411 yield* logManagedRelayAccountChange;485 let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout);
412 return false;486 for (;;) {
413 }487 const remaining = deadline - (yield* Clock.monotonicTimeNanos);
414 if (next.reason === "application-active-reconnect") {488 const probeEvent = yield* Effect.raceAllFirst([
415 // Mobile operating systems commonly suspend sockets without489 Fiber.await(probe).pipe(
416 // delivering a close event. A long background resume deliberately490 Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })),
417 // replaces that lease and starts a fresh attempt without backoff.491 ),
418 return true;492 takeSignal.pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))),
419 }493 Effect.sleep(Duration.nanos(remaining > 0n ? remaining : 0n)).pipe(
420 if (next.reason === "application-active" || next.reason === "application-active-probe") {494 Effect.as({ _tag: "TimedOut" as const }),
421 const probe = yield* lease.session.probe.pipe(495 ),
422 Effect.timeoutOrElse({496 ]);
423 duration:497 if (probeEvent._tag === "TimedOut") {
424 next.reason === "application-active-probe"498 yield* Fiber.interrupt(probe);
425 ? MOBILE_CONNECTION_PROBE_TIMEOUT499 return yield* new ConnectionTransientError({
426 : CONNECTION_PROBE_TIMEOUT,500 reason: "timeout",
427 orElse: () =>501 detail: `${target.label} did not respond to a connection health check.`,
428 Effect.fail(502 });
429 new ConnectionTransientError({503 }
430 reason: "timeout",504 if (probeEvent._tag === "ProbeCompleted") {
431 detail: `${target.label} did not respond to a connection health check.`,505 if (Exit.isSuccess(probeEvent.exit)) {
432 }),506 yield* Ref.set(probeUnanswered, false);
433 ),
434 }),
435 Effect.forkChild,
436 );
437 for (;;) {
438 const probeEvent = yield* Effect.raceFirst(
439 Fiber.await(probe).pipe(
440 Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })),
441 ),
442 Queue.take(signals).pipe(
443 Effect.map((signal) => ({ _tag: "Signal" as const, signal })),
444 ),
445 );
446 if (probeEvent._tag === "ProbeCompleted") {
447 if (Exit.isFailure(probeEvent.exit)) {
448 yield* Ref.set(wakeProbeFailed, true);
449 }
450 yield* probeEvent.exit;
451 break;
452 }
453 switch (probeEvent.signal._tag) {
454 case "DisconnectRequested":
455 case "RetryRequested":
456 yield* Fiber.interrupt(probe);
457 return false;
458 case "NetworkChanged":
459 if (probeEvent.signal.network === "offline") {
460 yield* Fiber.interrupt(probe);
461 return false;
462 }
463 break;
464 case "Wakeup":
465 if (probeEvent.signal.reason === "application-active-reconnect") {
466 yield* Fiber.interrupt(probe);
467 return true;
468 }
469 if (
470 probeEvent.signal.reason === "credentials-changed" &&
471 target._tag === "RelayConnectionTarget"
472 ) {
473 yield* Fiber.interrupt(probe);
474 return false;
475 }
476 break;
477 case "ConnectRequested":
478 break;
479 }
480 }
481 }507 }
508 yield* probeEvent.exit;
482 break;509 break;
483 case "ConnectRequested":510 }
484 break;511 const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal);
512 if (endDuringProbe !== undefined) {
513 yield* Fiber.interrupt(probe);
514 return endDuringProbe === "reset";
515 }
516 // A retry or an offline report during a desktop foreground probe wants
517 // its quicker answer, so it shortens the running probe.
518 const signalTimeout = probeTimeoutFor(probeEvent.signal);
519 if (signalTimeout !== undefined) {
520 const signalDeadline =
521 (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout);
522 if (signalDeadline < deadline) deadline = signalDeadline;
523 }
485 }524 }
486 }525 }
487 });526 });
...@@ -491,6 +530,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -491,6 +530,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
491 generation: number,530 generation: number,
492 lastFailure: ConnectionAttemptError | null,531 lastFailure: ConnectionAttemptError | null,
493 pendingRetry: Option.Option<PendingRetryTrace>,532 pendingRetry: Option.Option<PendingRetryTrace>,
533 ignoreOffline: boolean,
494 ) {534 ) {
495 yield* SubscriptionRef.set(prepared, Option.none());535 yield* SubscriptionRef.set(prepared, Option.none());
496 const establishment = yield* Effect.raceAllFirst([536 const establishment = yield* Effect.raceAllFirst([
...@@ -556,7 +596,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -556,7 +596,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
556596
557 const active = establishment.exit.value;597 const active = establishment.exit.value;
558 const currentIntent = yield* Ref.get(intent);598 const currentIntent = yield* Ref.get(intent);
559 if (!currentIntent.desired || currentIntent.network === "offline") {599 if (!currentIntent.desired || (currentIntent.network === "offline" && !ignoreOffline)) {
560 return {600 return {
561 _tag: "Interrupted",601 _tag: "Interrupted",
562 established: false,602 established: false,
...@@ -641,6 +681,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -641,6 +681,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
641 failureCount = 0;681 failureCount = 0;
642 pendingRetry = Option.none();682 pendingRetry = Option.none();
643 };683 };
684 // Set after a long resume ends an attempt or a session. The fresh attempt
685 // runs even while the network reports offline: the report is often wrong,
686 // and the replaced session must not leave the client offline.
687 let replacing = false;
644688
645 for (;;) {689 for (;;) {
646 if (yield* Ref.getAndSet(resetRetryState, false)) {690 if (yield* Ref.getAndSet(resetRetryState, false)) {
...@@ -657,7 +701,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -657,7 +701,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
657 yield* waitForSignal;701 yield* waitForSignal;
658 continue;702 continue;
659 }703 }
660 if (currentIntent.network === "offline") {704 if (currentIntent.network === "offline" && !replacing) {
661 yield* clearLease;705 yield* clearLease;
662 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));706 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));
663 const applicationActivated = yield* waitForSignal;707 const applicationActivated = yield* waitForSignal;
...@@ -670,11 +714,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -670,11 +714,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
670 const attempt = failureCount + 1;714 const attempt = failureCount + 1;
671 const nextGeneration = generation + 1;715 const nextGeneration = generation + 1;
672 const outcome: AttemptOutcome = yield* Effect.scoped(716 const outcome: AttemptOutcome = yield* Effect.scoped(
673 runAttempt(attempt, nextGeneration, latestFailure, pendingRetry),717 runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, replacing),
674 );718 );
719 replacing = false;
675 // Consumed on every iteration so a stale marker can never leak into a720 // Consumed on every iteration so a stale marker can never leak into a
676 // later, unrelated failure.721 // later, unrelated failure.
677 const failedWakeProbe = yield* Ref.getAndSet(wakeProbeFailed, false);722 const failedProbe = yield* Ref.getAndSet(probeUnanswered, false);
678 if (outcome.established) {723 if (outcome.established) {
679 generation = nextGeneration;724 generation = nextGeneration;
680 if (outcome.stable) {725 if (outcome.stable) {
...@@ -685,6 +730,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -685,6 +730,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
685 if (outcome._tag === "Interrupted") {730 if (outcome._tag === "Interrupted") {
686 if (outcome.resetRetry) {731 if (outcome.resetRetry) {
687 resetRetryLadder();732 resetRetryLadder();
733 replacing = true;
688 }734 }
689 continue;735 continue;
690 }736 }
...@@ -711,18 +757,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (...@@ -711,18 +757,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
711 continue;757 continue;
712 }758 }
713759
714 if (failedWakeProbe) {760 if (failedProbe) {
715 // The wake probe found a dead transport while the user is returning to761 // A probe found a dead transport, or the transport closed while a probe
716 // the app, so reconnect immediately instead of sleeping the first762 // waited for an answer (the user returned to the app, asked to retry,
717 // backoff rung. Only this first attempt skips the ladder; if it fails763 // or the network changed), so reconnect immediately instead
718 // too, normal backoff resumes.764 // of sleeping the first backoff rung. Only this first attempt skips the
765 // ladder; if it fails too, normal backoff resumes.
719 resetRetryLadder();766 resetRetryLadder();
720 yield* setState(connectingState(yield* Ref.get(intent), generation, 1, error));767 yield* setState(connectingState(yield* Ref.get(intent), generation, 1, error));
721 continue;768 continue;
722 }769 }
723770
724 failureCount += 1;771 failureCount += 1;
725 const delayMs = retryDelayMs(failureCount - 1);772 const delayMs = retryDelayMs(failureCount - 1, yield* Random.next);
726 pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({773 pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({
727 previousAttempt,774 previousAttempt,
728 failureCount,775 failureCount,
packages/client-runtime/src/connection/wakeups.ts+1
...@@ -16,6 +16,7 @@ export function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean {...@@ -16,6 +16,7 @@ export function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean {
16 );16 );
17}17}
1818
19// A long resume replaces the session, and the new session subscribes on its own.
19export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean {20export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean {
20 return reason === "application-active" || reason === "application-active-probe";21 return reason === "application-active" || reason === "application-active-probe";
21}22}
packages/client-runtime/src/state/server.ts+3-3
...@@ -179,9 +179,9 @@ export function validateServerUpdateReadyEvent(...@@ -179,9 +179,9 @@ export function validateServerUpdateReadyEvent(
179 * Keeps reconnect attempts ~1s apart for the whole update restart.179 * Keeps reconnect attempts ~1s apart for the whole update restart.
180 *180 *
181 * A restart takes the server down for ~15 seconds, but the supervisor's normal181 * A restart takes the server down for ~15 seconds, but the supervisor's normal
182 * backoff ladder (1/2/4/8/16s) assumes an unexpected failure and lands attempts182 * backoff assumes an unexpected failure and doubles its delay after each failed
183 * at ~3, 5, 9, 17 and 33 seconds — so a 15-second restart is observed as a183 * attempt, so a 15-second restart can be observed as a ~30-second "Resuming".
184 * 33-second "Resuming". Nudging on every backoff entry (not just the first)184 * Nudging on every backoff entry (not just the first)
185 * holds the retry cadence flat until the server answers again. The sleep before185 * holds the retry cadence flat until the server answers again. The sleep before
186 * each nudge is the pacer: a connection that fails instantly re-enters backoff186 * each nudge is the pacer: a connection that fails instantly re-enters backoff
187 * immediately and would otherwise spin a tight retry loop.187 * immediately and would otherwise spin a tight retry loop.