1import * as Cause from "effect/Cause";
2import * as Clock from "effect/Clock";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
8import * as Option from "effect/Option";
9import * as Queue from "effect/Queue";
10import * as Random from "effect/Random";
11import * as Ref from "effect/Ref";
12import * as Scope from "effect/Scope";
13import * as Stream from "effect/Stream";
14import * as SubscriptionRef from "effect/SubscriptionRef";
15import * as Tracer from "effect/Tracer";
16
17import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts";
18import * as Connectivity from "./connectivity.ts";
19import * as ConnectionDriver from "./driver.ts";
20import {
21 type ConnectionAttemptError,
22 type ConnectionTarget,
23 ConnectionTransientError,
24 type NetworkStatus,
25 type PreparedConnection,
26 type SupervisorConnectionState,
27} from "./model.ts";
28import * as RpcSession from "../rpc/session.ts";
29import { safeErrorLogAttributes } from "../errors/safeLog.ts";
30import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
31import * as ConnectionWakeups from "./wakeups.ts";
32import {
33 connectionRouteId,
34 connectionRoutes,
35 entryWithRoutes,
36 type ReportedEndpoint,
37} from "./routes.ts";
38
39const RETRY_BASE_DELAY_MS = 1_000;
40const RETRY_MAX_DELAY_MS = 300_000;
41// Per-route budget to resolve credentials and create the session, which ends
42// when the driver reports the synchronizing stage. Each route that gets that
43// far adds its own budget for the socket to open (bounded at 15 seconds by the
44// RPC session) and the server to deliver the first config snapshot. A cold
45// server with a large database can take longer than 15 seconds to answer, and
46// that is not a reason to drop a working connection.
47const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";
48const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT);
49const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds");
50const CONNECTION_PROBE_TIMEOUT = "15 seconds";
51// Mobile resumes, explicit retries, and offline events want a fast answer:
52// the user is waiting, or the network may be gone.
53const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds";
54const BACKOFF_RESET_AFTER_MS = 30_000;
55// While connected over a fallback route, how often to look for a better one.
56// Network changes and returning to the app also trigger a check.
57const BETTER_ROUTE_CHECK_INTERVAL = "60 seconds";
58// A better route that answered the check but then failed to connect is not
59// tried again for this long, so a flaky LAN cannot bounce the connection.
60const BETTER_ROUTE_COOLDOWN_MS = 300_000;
61
62interface SupervisorIntent {
63 readonly desired: boolean;
64 readonly network: NetworkStatus;
65}
66
67type SupervisorSignal =
68 | { readonly _tag: "ConnectRequested" }
69 | { readonly _tag: "DisconnectRequested" }
70 | { readonly _tag: "RetryRequested" }
71 | { readonly _tag: "NetworkChanged"; readonly network: NetworkStatus }
72 | { readonly _tag: "Wakeup"; readonly reason: ConnectionWakeups.ConnectionWakeup }
73 | { readonly _tag: "BetterRouteAvailable"; readonly routeId: string };
74
75interface PendingRetryTrace {
76 readonly previousAttempt: Tracer.Span;
77 readonly failureCount: number;
78 readonly delayMs: number;
79 readonly reason: ConnectionAttemptError["reason"];
80}
81
82interface TracedAttemptFailure {
83 readonly error: ConnectionAttemptError;
84 readonly attemptSpan: Option.Option<Tracer.Span>;
85}
86
87type AttemptOutcome =
88 | {
89 readonly _tag: "Interrupted";
90 readonly established: boolean;
91 readonly stable: boolean;
92 readonly resetRetry: boolean;
93 }
94 | {
95 readonly _tag: "Failure";
96 readonly established: boolean;
97 readonly stable: boolean;
98 readonly failure: TracedAttemptFailure;
99 };
100
101type EstablishmentEvent =
102 | {
103 readonly _tag: "Completed";
104 readonly exit: Exit.Exit<
105 {
106 readonly attemptSpan: Option.Option<Tracer.Span>;
107 readonly lease: ConnectionDriver.EnvironmentConnectionLease;
108 },
109 TracedAttemptFailure
110 >;
111 }
112 | { readonly _tag: "Interrupted"; readonly resetRetry: boolean }
113 | { readonly _tag: "TimedOut" };
114
115function exitUnlessInterrupted<A, E, R>(
116 effect: Effect.Effect<A, E, R>,
117): Effect.Effect<Exit.Exit<A, E>, never, R> {
118 return Effect.matchCauseEffect(effect, {
119 onFailure: (cause) =>
120 Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(Exit.failCause(cause)),
121 onSuccess: (value) => Effect.succeed(Exit.succeed(value)),
122 });
123}
124
125export interface EnvironmentSupervisorOptions {
126 readonly initiallyDesired?: boolean;
127 /**
128 * Saves the direct addresses the server reports once a session is ready and
129 * returns the entry with its updated routes, which later attempts use. The
130 * live session is left alone: it already works.
131 */
132 readonly learnRoutes?: (input: {
133 readonly activeRoute: ConnectionRoute;
134 readonly reported: ReadonlyArray<ReportedEndpoint>;
135 }) => Effect.Effect<Option.Option<ConnectionCatalogEntry>>;
136}
137
138/**
139 * Delay before the next attempt after `failureCount` consecutive failures
140 * (0 for the first retry). The ceiling doubles from 2s up to 5 minutes, and
141 * the delay is a random point in its upper half: never quicker than half the
142 * ceiling, and spread out so clients that lost the same server do not all
143 * reconnect in the same second. `random` is in [0, 1).
144 *
145 * The long cap only applies to a connection that keeps failing. Returning to
146 * the app, the network coming back, and an explicit retry all skip the wait.
147 */
148export function retryDelayMs(failureCount: number, random: number): number {
149 const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failureCount + 1));
150 return Math.round(ceiling / 2 + (ceiling / 2) * random);
151}
152
153function annotateTarget(target: ConnectionTarget) {
154 return Effect.annotateCurrentSpan({
155 "environment.id": target.environmentId,
156 "environment.label": target.label,
157 "environment.target.kind": target._tag,
158 });
159}
160
161function availableState(intent: SupervisorIntent, generation: number): SupervisorConnectionState {
162 return {
163 desired: false,
164 network: intent.network,
165 phase: "available",
166 stage: null,
167 attempt: 0,
168 generation,
169 lastFailure: null,
170 retryAt: null,
171 };
172}
173
174function offlineState(
175 intent: SupervisorIntent,
176 generation: number,
177 attempt: number,
178 lastFailure: ConnectionAttemptError | null,
179): SupervisorConnectionState {
180 return {
181 desired: true,
182 network: intent.network,
183 phase: "offline",
184 stage: null,
185 attempt,
186 generation,
187 lastFailure,
188 retryAt: null,
189 };
190}
191
192function connectingState(
193 intent: SupervisorIntent,
194 generation: number,
195 attempt: number,
196 lastFailure: ConnectionAttemptError | null,
197 stage: SupervisorConnectionState["stage"] = "preparing",
198): SupervisorConnectionState {
199 return {
200 desired: true,
201 network: intent.network,
202 phase: "connecting",
203 stage,
204 attempt,
205 generation,
206 lastFailure,
207 retryAt: null,
208 };
209}
210
211function failureFromExit<A>(
212 target: ConnectionTarget,
213 exit: Exit.Exit<A, TracedAttemptFailure>,
214 established: boolean,
215 stable: boolean,
216): AttemptOutcome {
217 if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) {
218 return { _tag: "Interrupted", established, stable, resetRetry: false };
219 }
220 const typedFailure = exit.cause.reasons.find(Cause.isFailReason);
221 if (typedFailure) {
222 return {
223 _tag: "Failure",
224 established,
225 stable,
226 failure: typedFailure.error,
227 };
228 }
229 return {
230 _tag: "Failure",
231 established,
232 stable,
233 failure: {
234 error: new ConnectionTransientError({
235 reason: "transport",
236 detail: `${target.label} connection failed unexpectedly.`,
237 }),
238 attemptSpan: Option.none(),
239 },
240 };
241}
242
243export class EnvironmentSupervisor extends Context.Service<
244 EnvironmentSupervisor,
245 {
246 readonly target: ConnectionTarget;
247 readonly state: SubscriptionRef.SubscriptionRef<SupervisorConnectionState>;
248 readonly session: SubscriptionRef.SubscriptionRef<Option.Option<RpcSession.RpcSession>>;
249 readonly prepared: SubscriptionRef.SubscriptionRef<Option.Option<PreparedConnection>>;
250 readonly connect: Effect.Effect<void>;
251 readonly disconnect: Effect.Effect<void>;
252 readonly retryNow: Effect.Effect<void>;
253 }
254>()("@t3tools/client-runtime/connection/supervisor/EnvironmentSupervisor") {}
255
256export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
257 entry: ConnectionCatalogEntry,
258 options?: EnvironmentSupervisorOptions,
259): Effect.fn.Return<
260 EnvironmentSupervisor["Service"],
261 never,
262 | Connectivity.Connectivity
263 | ConnectionDriver.ConnectionDriver
264 | Scope.Scope
265 | ConnectionWakeups.ConnectionWakeups
266> {
267 const target = entry.target;
268 // Relay-specific handling applies when any route is T3 Connect, since the
269 // attempt or the live session may be using it.
270 const usesRelay = connectionRoutes(entry).some(
271 (route) => route.target._tag === "RelayConnectionTarget",
272 );
273 const setupTimeoutDetail = `${target.label} did not respond during connection setup.${
274 usesRelay ? ` ${NETWORK_BLOCKING_HINT}` : ""
275 }`;
276 yield* annotateTarget(target);
277
278 const connectivity = yield* Connectivity.Connectivity;
279 const driver = yield* ConnectionDriver.ConnectionDriver;
280 const wakeups = yield* ConnectionWakeups.ConnectionWakeups;
281 const initialIntent: SupervisorIntent = {
282 desired: options?.initiallyDesired ?? false,
283 network: yield* connectivity.status,
284 };
285 const intent = yield* Ref.make(initialIntent);
286 const signals = yield* Queue.unbounded<SupervisorSignal>();
287 const resetRetryState = yield* Ref.make(false);
288 // Set while a probe of the live session is running, and kept when it fails
289 // or times out: something asked whether the connection still works and it
290 // closed or failed before answering, so the follow-up reconnect skips the
291 // first backoff rung instead of sleeping.
292 const probeUnanswered = yield* Ref.make(false);
293 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(
294 !initialIntent.desired
295 ? availableState(initialIntent, 0)
296 : initialIntent.network === "offline"
297 ? offlineState(initialIntent, 0, 0, null)
298 : connectingState(initialIntent, 0, 1, null),
299 );
300 const session = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(Option.none());
301 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
302 // Learned routes arrive while a session is live, so the route list is read
303 // fresh by each attempt and check instead of being fixed at creation.
304 const currentEntry = yield* Ref.make(entry);
305 const currentRoutes = Ref.get(currentEntry).pipe(Effect.map(connectionRoutes));
306 // Set when a better route answered while connected over a worse one; the
307 // replacement attempt tries it first. Cleared once an attempt starts.
308 const preferredRouteId = yield* Ref.make(Option.none<string>());
309 // Monotonic deadline per route that answered a check but failed to connect.
310 const routeCooldowns = yield* Ref.make<ReadonlyMap<string, bigint>>(new Map());
311
312 const attemptEntry = Effect.gen(function* () {
313 const latest = yield* Ref.get(currentEntry);
314 const routes = connectionRoutes(latest);
315 const preferred = yield* Ref.getAndSet(preferredRouteId, Option.none());
316 if (Option.isNone(preferred)) return latest;
317 const route = routes.find(
318 (candidate) => connectionRouteId(candidate.target) === preferred.value,
319 );
320 return route === undefined
321 ? latest
322 : entryWithRoutes(latest, [route, ...routes.filter((candidate) => candidate !== route)]);
323 });
324
325 const learnRoutesFrom = Effect.fnUntraced(function* (
326 learn: NonNullable<EnvironmentSupervisorOptions["learnRoutes"]>,
327 lease: ConnectionDriver.EnvironmentConnectionLease,
328 ) {
329 const config = yield* lease.session.initialConfig.pipe(Effect.option);
330 const reported = Option.getOrUndefined(config)?.directEndpoints;
331 if (reported === undefined) return;
332 const activeId = connectionRouteId(lease.prepared.target);
333 const activeRoute = (yield* currentRoutes).find(
334 (route) => connectionRouteId(route.target) === activeId,
335 );
336 if (activeRoute === undefined) return;
337 const updated = yield* learn({ activeRoute, reported });
338 if (Option.isSome(updated)) {
339 yield* Ref.set(currentEntry, updated.value);
340 // A learned route may rank above the one in use.
341 yield* requestBetterRouteCheck(lease);
342 }
343 });
344
345 const routeIndex = (lease: ConnectionDriver.EnvironmentConnectionLease) =>
346 currentRoutes.pipe(
347 Effect.map((routes) =>
348 routes.findIndex(
349 (route) => connectionRouteId(route.target) === connectionRouteId(lease.prepared.target),
350 ),
351 ),
352 );
353
354 /**
355 * Preflights the routes ranked above the one in use and signals the best
356 * that would connect. Routes without a cheap check (T3 Connect, SSH) never
357 * pass, so they are fallbacks, not destinations.
358 */
359 const checkBetterRoutes = Effect.fnUntraced(function* (
360 lease: ConnectionDriver.EnvironmentConnectionLease,
361 ) {
362 const latest = yield* Ref.get(currentEntry);
363 const current = yield* routeIndex(lease);
364 if (current <= 0) return;
365 const now = yield* Clock.monotonicTimeNanos;
366 const cooldowns = yield* Ref.get(routeCooldowns);
367 const better = connectionRoutes(latest)
368 .slice(0, current)
369 .filter((route) => (cooldowns.get(connectionRouteId(route.target)) ?? 0n) <= now);
370 const passed = yield* Effect.forEach(
371 better,
372 (route: ConnectionRoute) => driver.preflight(latest, route),
373 { concurrency: "unbounded" },
374 );
375 const index = passed.indexOf(true);
376 if (index === -1) return;
377 // The check may outlive the session it was asked about.
378 const live = yield* SubscriptionRef.get(session);
379 if (Option.isNone(live) || live.value !== lease.session) return;
380 yield* signal({
381 _tag: "BetterRouteAvailable",
382 routeId: connectionRouteId(better[index]!.target),
383 });
384 });
385
386 // One check at a time; a trigger during a check is dropped, not queued.
387 const betterRouteChecks = yield* Queue.dropping<ConnectionDriver.EnvironmentConnectionLease>(1);
388 const requestBetterRouteCheck = (lease: ConnectionDriver.EnvironmentConnectionLease) =>
389 routeIndex(lease).pipe(
390 Effect.flatMap((index) =>
391 index > 0 ? Queue.offer(betterRouteChecks, lease).pipe(Effect.asVoid) : Effect.void,
392 ),
393 );
394
395 const clearLease = Effect.all(
396 [SubscriptionRef.set(session, Option.none()), SubscriptionRef.set(prepared, Option.none())],
397 { discard: true },
398 );
399
400 const setState = Effect.fn("EnvironmentSupervisor.setState")(function* (
401 next: SupervisorConnectionState,
402 ) {
403 yield* SubscriptionRef.set(state, next);
404 });
405
406 const signal = Effect.fn("EnvironmentSupervisor.signal")(function* (next: SupervisorSignal) {
407 yield* Queue.offer(signals, next);
408 });
409
410 const logManagedRelayAccountChange = Effect.logInfo(
411 "Managed relay account changed; restarting the environment connection.",
412 ).pipe(
413 Effect.annotateLogs({
414 "environment.id": target.environmentId,
415 "environment.label": target.label,
416 }),
417 );
418
419 const reportProgress = Effect.fn("EnvironmentSupervisor.reportProgress")(function* (
420 attempt: number,
421 generation: number,
422 lastFailure: ConnectionAttemptError | null,
423 progress: ConnectionDriver.ConnectionDriverProgress,
424 ) {
425 if ("prepared" in progress) {
426 yield* SubscriptionRef.set(prepared, Option.some(progress.prepared));
427 }
428 yield* setState(
429 connectingState(yield* Ref.get(intent), generation, attempt, lastFailure, progress.stage),
430 );
431 });
432
433 const establishConnection = Effect.fnUntraced(function* (
434 attempt: number,
435 generation: number,
436 lastFailure: ConnectionAttemptError | null,
437 synchronizedRoutes: Ref.Ref<number>,
438 ) {
439 return yield* driver.connect(yield* attemptEntry, (progress) =>
440 (progress.stage === "synchronizing"
441 ? Ref.update(synchronizedRoutes, (count) => count + 1)
442 : Effect.void
443 ).pipe(Effect.andThen(reportProgress(attempt, generation, lastFailure, progress))),
444 );
445 });
446
447 const traceRelayEstablishment = (
448 effect: Effect.Effect<
449 ConnectionDriver.EnvironmentConnectionLease,
450 ConnectionAttemptError,
451 Scope.Scope
452 >,
453 attempt: number,
454 generation: number,
455 pendingRetry: Option.Option<PendingRetryTrace>,
456 ) => {
457 const traced = Effect.gen(function* () {
458 const attemptSpan = yield* Effect.currentSpan.pipe(Effect.orDie);
459 yield* annotateTarget(target);
460 yield* Effect.annotateCurrentSpan({
461 "connection.attempt": attempt,
462 "connection.generation": generation,
463 "connection.retry.failure_count": Option.match(pendingRetry, {
464 onNone: () => 0,
465 onSome: (retry) => retry.failureCount,
466 }),
467 });
468 const lease = yield* effect.pipe(
469 Effect.mapError((error): TracedAttemptFailure => ({
470 error,
471 attemptSpan: Option.some(attemptSpan),
472 })),
473 );
474 return { attemptSpan: Option.some(attemptSpan), lease };
475 }).pipe(Effect.withSpan("relay.connection.attempt", { root: true }));
476
477 return Option.match(pendingRetry, {
478 onNone: () => traced,
479 onSome: (retry) =>
480 traced.pipe(
481 Effect.linkSpans(retry.previousAttempt, {
482 "connection.retry.delay_ms": retry.delayMs,
483 "connection.retry.reason": retry.reason,
484 }),
485 ),
486 });
487 };
488
489 const establishTracedConnection = Effect.fnUntraced(function* (
490 attempt: number,
491 generation: number,
492 lastFailure: ConnectionAttemptError | null,
493 pendingRetry: Option.Option<PendingRetryTrace>,
494 synchronizedRoutes: Ref.Ref<number>,
495 ) {
496 if (usesRelay) {
497 return yield* traceRelayEstablishment(
498 establishConnection(attempt, generation, lastFailure, synchronizedRoutes),
499 attempt,
500 generation,
501 pendingRetry,
502 );
503 }
504 return yield* establishConnection(attempt, generation, lastFailure, synchronizedRoutes).pipe(
505 Effect.map((lease) => ({
506 attemptSpan: Option.none<Tracer.Span>(),
507 lease,
508 })),
509 Effect.mapError((error): TracedAttemptFailure => ({
510 error,
511 attemptSpan: Option.none(),
512 })),
513 );
514 });
515
516 const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () {
517 for (;;) {
518 const next = yield* Queue.take(signals);
519 switch (next._tag) {
520 case "DisconnectRequested":
521 case "RetryRequested":
522 return false;
523 case "NetworkChanged":
524 if (next.network === "offline") {
525 return false;
526 }
527 break;
528 case "ConnectRequested":
529 case "BetterRouteAvailable":
530 break;
531 case "Wakeup":
532 if (next.reason === "application-active-reconnect") {
533 return true;
534 }
535 if (next.reason === "credentials-changed" && usesRelay) {
536 yield* logManagedRelayAccountChange;
537 return false;
538 }
539 break;
540 }
541 }
542 });
543
544 // Signals that end a connected lease whatever its health: "reset" ends it
545 // and restarts the retry ladder, "end" ends it, undefined keeps it.
546 const isRelayLease = (lease: ConnectionDriver.EnvironmentConnectionLease) =>
547 lease.prepared.target._tag === "RelayConnectionTarget";
548
549 const connectedLeaseEnd = Effect.fnUntraced(function* (
550 next: SupervisorSignal,
551 lease: ConnectionDriver.EnvironmentConnectionLease,
552 ) {
553 if (next._tag === "DisconnectRequested") {
554 return "end" as const;
555 }
556 if (next._tag === "BetterRouteAvailable") {
557 // Replaced like a long resume: the new attempt prefers the better route
558 // and its session takes over the durable subscriptions.
559 yield* Ref.set(preferredRouteId, Option.some(next.routeId));
560 return "reset" as const;
561 }
562 if (next._tag !== "Wakeup") {
563 return undefined;
564 }
565 if (next.reason === "application-active-reconnect") {
566 // Mobile operating systems often kill a suspended socket without a close
567 // event. A probe would show a dead socket as "Resuming" until it times
568 // out, so a long background resume replaces the session at once.
569 return "reset" as const;
570 }
571 // Only a session over T3 Connect holds the old account's credential.
572 if (next.reason === "credentials-changed" && isRelayLease(lease)) {
573 yield* logManagedRelayAccountChange;
574 return "end" as const;
575 }
576 return undefined;
577 });
578
579 // How long a signal waits for the live session to answer a probe, or
580 // undefined when the signal does not question the connection.
581 const probeTimeoutFor = (next: SupervisorSignal): Duration.Input | undefined => {
582 switch (next._tag) {
583 case "RetryRequested":
584 return QUICK_CONNECTION_PROBE_TIMEOUT;
585 case "NetworkChanged":
586 return next.network === "offline" ? QUICK_CONNECTION_PROBE_TIMEOUT : undefined;
587 case "Wakeup":
588 if (next.reason === "application-active") {
589 return CONNECTION_PROBE_TIMEOUT;
590 }
591 // A socket opened on the previous network may now be unroutable.
592 return next.reason === "application-active-probe" || next.reason === "network-changed"
593 ? QUICK_CONNECTION_PROBE_TIMEOUT
594 : undefined;
595 case "ConnectRequested":
596 case "DisconnectRequested":
597 case "BetterRouteAvailable":
598 return undefined;
599 }
600 };
601
602 // Holds a connected lease until it must end, and returns whether to restart
603 // the retry ladder. Returning to the app, an explicit retry, and the network
604 // reporting offline all probe the live session instead of replacing it, so a
605 // healthy socket is not torn down (the offline report is often wrong, for
606 // example for a loopback server). Only a long mobile resume replaces the
607 // session without a probe. A failed probe fails this effect, and the
608 // supervisor reconnects.
609 const monitorConnectedLease = Effect.fnUntraced(function* (
610 lease: ConnectionDriver.EnvironmentConnectionLease,
611 ) {
612 // A probe answers an explicit retry here, so the retry must not also reset
613 // the backoff of a later, unrelated failure.
614 const takeSignal = Queue.take(signals).pipe(
615 Effect.tap((next) =>
616 next._tag === "RetryRequested" ? Ref.set(resetRetryState, false) : Effect.void,
617 ),
618 );
619 // Ticks always run: the check skips the preferred route itself, and the
620 // route list can grow while connected (learned routes).
621 yield* Stream.tick(BETTER_ROUTE_CHECK_INTERVAL).pipe(
622 Stream.drop(1),
623 Stream.runForEach(() => requestBetterRouteCheck(lease)),
624 Effect.forkScoped,
625 );
626 for (;;) {
627 const next = yield* takeSignal;
628 const end = yield* connectedLeaseEnd(next, lease);
629 if (end !== undefined) {
630 return end === "reset";
631 }
632 // A new network or a return to the app may have brought a better route back.
633 if (
634 (next._tag === "NetworkChanged" && next.network !== "offline") ||
635 (next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason))
636 ) {
637 yield* requestBetterRouteCheck(lease);
638 }
639 const probeTimeout = probeTimeoutFor(next);
640 if (probeTimeout === undefined) {
641 continue;
642 }
643 yield* Ref.set(probeUnanswered, true);
644 const probe = yield* Effect.forkChild(lease.session.probe);
645 // Monotonic nanoseconds, so a wall-clock correction cannot move the deadline.
646 let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout);
647 for (;;) {
648 const remaining = deadline - (yield* Clock.monotonicTimeNanos);
649 const probeEvent = yield* Effect.raceAllFirst([
650 Fiber.await(probe).pipe(
651 Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })),
652 ),
653 takeSignal.pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))),
654 Effect.sleep(Duration.nanos(remaining > 0n ? remaining : 0n)).pipe(
655 Effect.as({ _tag: "TimedOut" as const }),
656 ),
657 ]);
658 if (probeEvent._tag === "TimedOut") {
659 yield* Fiber.interrupt(probe);
660 return yield* new ConnectionTransientError({
661 reason: "timeout",
662 detail: `${target.label} did not respond to a connection health check.`,
663 });
664 }
665 if (probeEvent._tag === "ProbeCompleted") {
666 if (Exit.isSuccess(probeEvent.exit)) {
667 yield* Ref.set(probeUnanswered, false);
668 }
669 yield* probeEvent.exit;
670 break;
671 }
672 const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease);
673 if (endDuringProbe !== undefined) {
674 yield* Fiber.interrupt(probe);
675 return endDuringProbe === "reset";
676 }
677 // A retry or an offline report during a desktop foreground probe wants
678 // its quicker answer, so it shortens the running probe.
679 const signalTimeout = probeTimeoutFor(probeEvent.signal);
680 if (signalTimeout !== undefined) {
681 const signalDeadline =
682 (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout);
683 if (signalDeadline < deadline) deadline = signalDeadline;
684 }
685 }
686 }
687 });
688
689 const runAttempt = Effect.fnUntraced(function* (
690 attempt: number,
691 generation: number,
692 lastFailure: ConnectionAttemptError | null,
693 pendingRetry: Option.Option<PendingRetryTrace>,
694 ignoreOffline: boolean,
695 ) {
696 const switchingTo = yield* Ref.get(preferredRouteId);
697 yield* SubscriptionRef.set(prepared, Option.none());
698 const synchronizedRoutes = yield* Ref.make(0);
699 const establishment = yield* Effect.raceAllFirst([
700 exitUnlessInterrupted(
701 establishTracedConnection(
702 attempt,
703 generation,
704 lastFailure,
705 pendingRetry,
706 synchronizedRoutes,
707 ),
708 ).pipe(
709 Effect.map((exit): EstablishmentEvent => ({
710 _tag: "Completed",
711 exit,
712 })),
713 ),
714 waitForEstablishmentInterrupt().pipe(
715 Effect.map((resetRetry): EstablishmentEvent => ({
716 _tag: "Interrupted",
717 resetRetry,
718 })),
719 ),
720 // Each route may use the full setup time before the next is tried, and
721 // each route that reaches synchronizing adds its own snapshot budget, so
722 // a route that fails late does not eat the next route's time.
723 Effect.gen(function* () {
724 const routeCount = connectionRoutes(yield* Ref.get(currentEntry)).length;
725 const startedAt = yield* Clock.monotonicTimeNanos;
726 const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routeCount));
727 while (true) {
728 const synchronized = BigInt(yield* Ref.get(synchronizedRoutes));
729 const deadline =
730 startedAt + setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout);
731 const remaining = deadline - (yield* Clock.monotonicTimeNanos);
732 if (remaining <= 0n) {
733 return { _tag: "TimedOut" } as const satisfies EstablishmentEvent;
734 }
735 yield* Effect.sleep(Duration.nanos(remaining));
736 }
737 }),
738 ]);
739
740 if (establishment._tag === "Interrupted") {
741 return {
742 _tag: "Interrupted",
743 established: false,
744 stable: false,
745 resetRetry: establishment.resetRetry,
746 } satisfies AttemptOutcome;
747 }
748 if (establishment._tag === "TimedOut") {
749 return {
750 _tag: "Failure",
751 established: false,
752 stable: false,
753 failure: {
754 error: new ConnectionTransientError({
755 reason: "timeout",
756 detail: setupTimeoutDetail,
757 }),
758 attemptSpan: Option.none(),
759 },
760 } satisfies AttemptOutcome;
761 }
762 if (Exit.isFailure(establishment.exit)) {
763 const isUnexpectedDefect =
764 !Cause.hasInterruptsOnly(establishment.exit.cause) &&
765 !establishment.exit.cause.reasons.some(Cause.isFailReason);
766 const outcome = failureFromExit(target, establishment.exit, false, false);
767 if (isUnexpectedDefect) {
768 const defect = establishment.exit.cause.reasons.find(Cause.isDieReason)?.defect;
769 yield* Effect.logError("Connection attempt failed with an unexpected defect.").pipe(
770 Effect.annotateLogs({
771 "environment.id": target.environmentId,
772 "environment.label": target.label,
773 "cause.reason_count": establishment.exit.cause.reasons.length,
774 ...safeErrorLogAttributes(defect),
775 }),
776 );
777 }
778 return outcome;
779 }
780
781 const active = establishment.exit.value;
782 if (Option.isSome(switchingTo)) {
783 const landed = connectionRouteId(active.lease.prepared.target);
784 if (landed !== switchingTo.value) {
785 const until =
786 (yield* Clock.monotonicTimeNanos) + BigInt(BETTER_ROUTE_COOLDOWN_MS) * 1_000_000n;
787 yield* Ref.update(routeCooldowns, (current) =>
788 new Map(current).set(switchingTo.value, until),
789 );
790 }
791 }
792 const currentIntent = yield* Ref.get(intent);
793 if (!currentIntent.desired || (currentIntent.network === "offline" && !ignoreOffline)) {
794 return {
795 _tag: "Interrupted",
796 established: false,
797 stable: false,
798 resetRetry: false,
799 } satisfies AttemptOutcome;
800 }
801
802 const connectedAt = yield* Clock.currentTimeMillis;
803 yield* SubscriptionRef.set(prepared, Option.some(active.lease.prepared));
804 yield* SubscriptionRef.set(session, Option.some(active.lease.session));
805 if (options?.learnRoutes !== undefined) {
806 yield* learnRoutesFrom(options.learnRoutes, active.lease).pipe(Effect.forkScoped);
807 }
808 yield* setState({
809 desired: true,
810 network: currentIntent.network,
811 phase: "connected",
812 stage: null,
813 attempt,
814 generation,
815 lastFailure: null,
816 retryAt: null,
817 });
818
819 const connectedExit = yield* Effect.raceFirst(
820 active.lease.session.closed.pipe(
821 Effect.mapError((error): TracedAttemptFailure => ({
822 error,
823 attemptSpan: active.attemptSpan,
824 })),
825 ),
826 monitorConnectedLease(active.lease).pipe(
827 Effect.mapError((error): TracedAttemptFailure => ({
828 error,
829 attemptSpan: active.attemptSpan,
830 })),
831 ),
832 ).pipe(exitUnlessInterrupted);
833 const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt;
834 if (Exit.isSuccess(connectedExit)) {
835 return {
836 _tag: "Interrupted",
837 established: true,
838 stable: connectedForMs >= BACKOFF_RESET_AFTER_MS,
839 resetRetry: connectedExit.value,
840 } satisfies AttemptOutcome;
841 }
842 const outcome = failureFromExit(
843 target,
844 connectedExit,
845 true,
846 connectedForMs >= BACKOFF_RESET_AFTER_MS,
847 );
848 if (outcome._tag === "Failure") {
849 // A live session ending is otherwise invisible in the client trace, so
850 // record why, and how long it lasted, as its own root span.
851 yield* Effect.void.pipe(
852 Effect.withSpan("EnvironmentSupervisor.connectionLost", {
853 root: true,
854 attributes: {
855 "environment.id": target.environmentId,
856 "environment.label": target.label,
857 "environment.target.kind": target._tag,
858 "connection.connected_ms": connectedForMs,
859 "connection.failure.reason": outcome.failure.error.reason,
860 "connection.failure.detail": outcome.failure.error.detail,
861 },
862 }),
863 );
864 }
865 return outcome;
866 }, Effect.ensuring(clearLease));
867
868 const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) {
869 // @effect-diagnostics-next-line raceFirstWithSleepToTimeout:off - the sleep is the retry delay (false), not a timeout around the signal loop
870 return yield* Effect.raceFirst(
871 Effect.sleep(delayMs).pipe(Effect.as(false)),
872 Effect.gen(function* () {
873 for (;;) {
874 const next = yield* Queue.take(signals);
875 switch (next._tag) {
876 case "Wakeup":
877 return ConnectionWakeups.resetsRetryBackoff(next.reason);
878 case "ConnectRequested":
879 case "DisconnectRequested":
880 case "RetryRequested":
881 case "NetworkChanged":
882 return false;
883 case "BetterRouteAvailable":
884 break;
885 }
886 }
887 }),
888 );
889 });
890
891 // A better route only matters to a live session, so idle states ignore it.
892 const waitForSignal = Queue.take(signals).pipe(
893 Effect.repeat({ while: (next) => next._tag === "BetterRouteAvailable" }),
894 Effect.map(
895 (next) => next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason),
896 ),
897 );
898
899 const run = Effect.fnUntraced(function* () {
900 let failureCount = 0;
901 let generation = 0;
902 let latestFailure: ConnectionAttemptError | null = null;
903 let pendingRetry = Option.none<PendingRetryTrace>();
904 const resetRetryLadder = () => {
905 failureCount = 0;
906 pendingRetry = Option.none();
907 };
908 // Set after a long resume ends an attempt or a session. The fresh attempt
909 // runs even while the network reports offline: the report is often wrong,
910 // and the replaced session must not leave the client offline.
911 let replacing = false;
912
913 for (;;) {
914 if (yield* Ref.getAndSet(resetRetryState, false)) {
915 failureCount = 0;
916 latestFailure = null;
917 pendingRetry = Option.none();
918 }
919 const currentIntent = yield* Ref.get(intent);
920 if (!currentIntent.desired) {
921 resetRetryLadder();
922 latestFailure = null;
923 yield* clearLease;
924 yield* setState(availableState(currentIntent, generation));
925 yield* waitForSignal;
926 continue;
927 }
928 if (currentIntent.network === "offline" && !replacing) {
929 yield* clearLease;
930 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));
931 const applicationActivated = yield* waitForSignal;
932 if (applicationActivated) {
933 resetRetryLadder();
934 }
935 continue;
936 }
937
938 const attempt = failureCount + 1;
939 const nextGeneration = generation + 1;
940 const outcome: AttemptOutcome = yield* Effect.scoped(
941 runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, replacing),
942 );
943 replacing = false;
944 // Consumed on every iteration so a stale marker can never leak into a
945 // later, unrelated failure.
946 const failedProbe = yield* Ref.getAndSet(probeUnanswered, false);
947 if (outcome.established) {
948 generation = nextGeneration;
949 if (outcome.stable) {
950 resetRetryLadder();
951 latestFailure = null;
952 }
953 }
954 if (outcome._tag === "Interrupted") {
955 if (outcome.resetRetry) {
956 resetRetryLadder();
957 replacing = true;
958 }
959 continue;
960 }
961
962 const attemptSpan: Option.Option<Tracer.Span> = outcome.failure.attemptSpan;
963 const error: ConnectionAttemptError = outcome.failure.error;
964 latestFailure = error;
965 if (error._tag === "ConnectionBlockedError") {
966 const blockedIntent = yield* Ref.get(intent);
967 yield* setState({
968 desired: blockedIntent.desired,
969 network: blockedIntent.network,
970 phase: "blocked",
971 stage: null,
972 attempt,
973 generation,
974 lastFailure: error,
975 retryAt: null,
976 });
977 const applicationActivated = yield* waitForSignal;
978 if (applicationActivated) {
979 resetRetryLadder();
980 }
981 continue;
982 }
983
984 if (failedProbe) {
985 // A probe found a dead transport, or the transport closed while a probe
986 // waited for an answer (the user returned to the app, asked to retry,
987 // or the network changed), so reconnect immediately instead
988 // of sleeping the first backoff rung. Only this first attempt skips the
989 // ladder; if it fails too, normal backoff resumes.
990 resetRetryLadder();
991 yield* setState(connectingState(yield* Ref.get(intent), generation, 1, error));
992 continue;
993 }
994
995 failureCount += 1;
996 const delayMs = retryDelayMs(failureCount - 1, yield* Random.next);
997 pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({
998 previousAttempt,
999 failureCount,
1000 delayMs,
1001 reason: error.reason,
1002 }));
1003 const failedIntent = yield* Ref.get(intent);
1004 yield* setState({
1005 desired: failedIntent.desired,
1006 network: failedIntent.network,
1007 phase: "backoff",
1008 stage: null,
1009 attempt,
1010 generation,
1011 lastFailure: error,
1012 retryAt: (yield* Clock.currentTimeMillis) + delayMs,
1013 });
1014 const applicationActivated = yield* waitForRetrySignal(delayMs);
1015 if (applicationActivated) {
1016 resetRetryLadder();
1017 }
1018 }
1019 });
1020
1021 yield* connectivity.changes.pipe(
1022 Stream.runForEach((network) =>
1023 Ref.modify(intent, (current) =>
1024 current.network === network ? [false, current] : ([true, { ...current, network }] as const),
1025 ).pipe(
1026 Effect.flatMap((changed) =>
1027 changed ? signal({ _tag: "NetworkChanged", network }) : Effect.void,
1028 ),
1029 ),
1030 ),
1031 Effect.forkScoped,
1032 );
1033 yield* wakeups.changes.pipe(
1034 Stream.runForEach((reason) => signal({ _tag: "Wakeup", reason })),
1035 Effect.forkScoped,
1036 );
1037 yield* Queue.take(betterRouteChecks).pipe(
1038 Effect.flatMap(checkBetterRoutes),
1039 Effect.forever,
1040 Effect.forkScoped,
1041 );
1042 yield* run().pipe(Effect.forkScoped);
1043
1044 const connect = Ref.update(intent, (current) => ({
1045 ...current,
1046 desired: true,
1047 })).pipe(
1048 Effect.andThen(signal({ _tag: "ConnectRequested" })),
1049 Effect.withSpan("EnvironmentSupervisor.connect"),
1050 );
1051
1052 const disconnect = Ref.update(intent, (current) => ({
1053 ...current,
1054 desired: false,
1055 })).pipe(
1056 Effect.andThen(signal({ _tag: "DisconnectRequested" })),
1057 Effect.withSpan("EnvironmentSupervisor.disconnect"),
1058 );
1059
1060 const retryNow = Ref.set(resetRetryState, true).pipe(
1061 Effect.andThen(signal({ _tag: "RetryRequested" })),
1062 Effect.withSpan("EnvironmentSupervisor.retryNow"),
1063 );
1064
1065 yield* Effect.addFinalizer(() => Queue.shutdown(signals).pipe(Effect.andThen(clearLease)));
1066
1067 return EnvironmentSupervisor.of({
1068 target,
1069 state,
1070 session,
1071 prepared,
1072 connect,
1073 disconnect,
1074 retryNow,
1075 });
1076});