Open History to see the last commit
1import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
2import * as Cause from "effect/Cause";
3import * as Clock from "effect/Clock";
4import * as Context from "effect/Context";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Fiber from "effect/Fiber";
9import * as Option from "effect/Option";
10import * as Queue from "effect/Queue";
11import * as Random from "effect/Random";
12import * as Ref from "effect/Ref";
13import * as Scope from "effect/Scope";
14import * as Stream from "effect/Stream";
15import * as SubscriptionRef from "effect/SubscriptionRef";
16import * as Tracer from "effect/Tracer";
17
18import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts";
19import * as Connectivity from "./connectivity.ts";
20import * as ConnectionDriver from "./driver.ts";
21import {
22 type ConnectionAttemptError,
23 type ConnectionTarget,
24 ConnectionTransientError,
25 type NetworkStatus,
26 type PreparedConnection,
27 type SupervisorConnectionState,
28} from "./model.ts";
29import * as RpcSession from "../rpc/session.ts";
30import { safeErrorLogAttributes } from "../errors/safeLog.ts";
31import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
32import * as ConnectionWakeups from "./wakeups.ts";
33import {
34 connectionRouteId,
35 connectionRouteKind,
36 connectionRoutes,
37 type ConnectionRouteKind,
38 entryWithRoutes,
39 insertRoute,
40 type ReportedEndpoint,
41} from "./routes.ts";
42
43const RETRY_BASE_DELAY_MS = 1_000;
44const RETRY_MAX_DELAY_MS = 300_000;
45const SESSION_LOSS_RETRY_DELAY_MS = 250;
46// A cold server with a large database can take longer than 15 seconds to
47// answer, and that is not a reason to drop a working connection.
48const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";
49const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT);
50const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds");
51const CONNECTION_PROBE_TIMEOUT = "15 seconds";
52const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds";
53// After a long suspension pooled connections are often dead too, so the host
54// check can hang with the socket. A replacement is cheaper than the wait.
55const LONG_RESUME_PROBE_TIMEOUT = "1500 millis";
56// A probe still unanswered this long checks the host with a fresh request. If
57// the host answers while the socket stays silent, the socket is dead.
58const SILENT_SOCKET_CHECK_AFTER = "500 millis";
59const SILENT_SOCKET_RECHECK_EVERY = "1 second";
60const FRAME_CHECK_EVERY = "100 millis";
61// Off the LAN one lost packet can stall a live socket for a second, so a remote
62// socket must stay silent this long, except after a long resume. It must end
63// before QUICK_CONNECTION_PROBE_TIMEOUT so a resume replaces a silent socket
64// instead of timing out.
65const REMOTE_SILENT_SOCKET_AFTER = "2500 millis";
66const BACKOFF_RESET_AFTER_MS = 30_000;
67const BETTER_ROUTE_CHECK_INTERVAL_MS = 60_000;
68// Each check of a dead route can wait out the full probe timeout, so a route
69// that keeps failing is checked half as often each time, until the network
70// changes and it may answer again.
71const SILENT_ROUTE_BACKOFF_MAX_MS = 1_800_000;
72// A better route that answered the check but then failed to connect is not
73// tried again for this long, so a flaky LAN cannot bounce the connection.
74const BETTER_ROUTE_COOLDOWN_MS = 300_000;
75
76interface SupervisorIntent {
77 readonly desired: boolean;
78 readonly network: NetworkStatus;
79}
80
81type SupervisorSignal =
82 | { readonly _tag: "ConnectRequested" }
83 | { readonly _tag: "DisconnectRequested" }
84 | { readonly _tag: "RetryRequested" }
85 | { readonly _tag: "NetworkChanged"; readonly network: NetworkStatus }
86 | { readonly _tag: "Wakeup"; readonly reason: ConnectionWakeups.ConnectionWakeup }
87 | { readonly _tag: "BetterRouteAvailable"; readonly routeId: string };
88
89interface PendingRetryTrace {
90 readonly previousAttempt: Tracer.Span;
91 readonly failureCount: number;
92 readonly delayMs: number;
93 readonly reason: ConnectionAttemptError["reason"];
94}
95
96type ConnectedLease = ConnectionDriver.EnvironmentConnectionLease & {
97 readonly activeRoute: ConnectionRoute | undefined;
98};
99
100interface TracedAttemptFailure {
101 readonly error: ConnectionAttemptError;
102 readonly attemptSpan: Option.Option<Tracer.Span>;
103}
104
105type AttemptOutcome =
106 | {
107 readonly _tag: "Interrupted";
108 readonly established: boolean;
109 readonly stable: boolean;
110 readonly resetRetry: boolean;
111 }
112 | {
113 readonly _tag: "Failure";
114 readonly established: boolean;
115 readonly stable: boolean;
116 readonly failure: TracedAttemptFailure;
117 };
118
119type EstablishmentEvent =
120 | {
121 readonly _tag: "Completed";
122 readonly exit: Exit.Exit<
123 {
124 readonly attemptSpan: Option.Option<Tracer.Span>;
125 readonly lease: ConnectionDriver.EnvironmentConnectionLease;
126 },
127 TracedAttemptFailure
128 >;
129 }
130 | { readonly _tag: "Interrupted"; readonly resetRetry: boolean }
131 | { readonly _tag: "TimedOut" };
132
133function exitUnlessInterrupted<A, E, R>(
134 effect: Effect.Effect<A, E, R>,
135): Effect.Effect<Exit.Exit<A, E>, never, R> {
136 return Effect.matchCauseEffect(effect, {
137 onFailure: (cause) =>
138 Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(Exit.failCause(cause)),
139 onSuccess: (value) => Effect.succeed(Exit.succeed(value)),
140 });
141}
142
143export interface EnvironmentSupervisorOptions {
144 readonly initiallyDesired?: boolean;
145 readonly learnRoutes?: (input: {
146 readonly activeRoute: ConnectionRoute;
147 readonly reported: ReadonlyArray<ReportedEndpoint>;
148 }) => Effect.Effect<Option.Option<ConnectionCatalogEntry>>;
149}
150
151function retryDelayMs(failureCount: number, random: number): number {
152 const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failureCount + 1));
153 return Math.round(ceiling / 2 + (ceiling / 2) * random);
154}
155
156interface SilentRoute {
157 readonly failures: number;
158 readonly until: bigint;
159}
160
161function backOffSilentRoute(previous: SilentRoute | undefined, now: bigint): SilentRoute {
162 const failures = (previous?.failures ?? 0) + 1;
163 const delayMs = Math.min(
164 SILENT_ROUTE_BACKOFF_MAX_MS,
165 BETTER_ROUTE_CHECK_INTERVAL_MS * 2 ** failures,
166 );
167 return { failures, until: now + BigInt(delayMs) * 1_000_000n };
168}
169
170function annotateTarget(target: ConnectionTarget) {
171 return Effect.annotateCurrentSpan({
172 "environment.id": target.environmentId,
173 "environment.label": target.label,
174 "environment.target.kind": target._tag,
175 });
176}
177
178function availableState(intent: SupervisorIntent, generation: number): SupervisorConnectionState {
179 return {
180 desired: false,
181 network: intent.network,
182 phase: "available",
183 stage: null,
184 attempt: 0,
185 generation,
186 lastFailure: null,
187 retryAt: null,
188 };
189}
190
191function offlineState(
192 intent: SupervisorIntent,
193 generation: number,
194 attempt: number,
195 lastFailure: ConnectionAttemptError | null,
196): SupervisorConnectionState {
197 return {
198 desired: true,
199 network: intent.network,
200 phase: "offline",
201 stage: null,
202 attempt,
203 generation,
204 lastFailure,
205 retryAt: null,
206 };
207}
208
209function connectingState(
210 intent: SupervisorIntent,
211 generation: number,
212 attempt: number,
213 lastFailure: ConnectionAttemptError | null,
214 stage: SupervisorConnectionState["stage"] = "preparing",
215): SupervisorConnectionState {
216 return {
217 desired: true,
218 network: intent.network,
219 phase: "connecting",
220 stage,
221 attempt,
222 generation,
223 lastFailure,
224 retryAt: null,
225 };
226}
227
228function failureFromExit<A>(
229 target: ConnectionTarget,
230 exit: Exit.Exit<A, TracedAttemptFailure>,
231 established: boolean,
232 stable: boolean,
233): AttemptOutcome {
234 if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) {
235 return { _tag: "Interrupted", established, stable, resetRetry: false };
236 }
237 const typedFailure = exit.cause.reasons.find(Cause.isFailReason);
238 if (typedFailure) {
239 return {
240 _tag: "Failure",
241 established,
242 stable,
243 failure: typedFailure.error,
244 };
245 }
246 return {
247 _tag: "Failure",
248 established,
249 stable,
250 failure: {
251 error: new ConnectionTransientError({
252 reason: "transport",
253 detail: `${target.label} connection failed unexpectedly.`,
254 }),
255 attemptSpan: Option.none(),
256 },
257 };
258}
259
260export class EnvironmentSupervisor extends Context.Service<
261 EnvironmentSupervisor,
262 {
263 readonly target: ConnectionTarget;
264 readonly state: SubscriptionRef.SubscriptionRef<SupervisorConnectionState>;
265 readonly session: SubscriptionRef.SubscriptionRef<Option.Option<RpcSession.RpcSession>>;
266 readonly prepared: SubscriptionRef.SubscriptionRef<Option.Option<PreparedConnection>>;
267 readonly connect: Effect.Effect<void>;
268 readonly disconnect: Effect.Effect<void>;
269 readonly retryNow: Effect.Effect<void>;
270 }
271>()("@t3tools/client-runtime/connection/supervisor/EnvironmentSupervisor") {}
272
273export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
274 entry: ConnectionCatalogEntry,
275 options?: EnvironmentSupervisorOptions,
276): Effect.fn.Return<
277 EnvironmentSupervisor["Service"],
278 never,
279 | Connectivity.Connectivity
280 | ConnectionDriver.ConnectionDriver
281 | Scope.Scope
282 | ConnectionWakeups.ConnectionWakeups
283> {
284 const target = entry.target;
285 const usesRelay = connectionRoutes(entry).some(
286 (route) => route.target._tag === "RelayConnectionTarget",
287 );
288 const setupTimeoutDetail = `${target.label} did not respond during connection setup.${
289 usesRelay ? ` ${NETWORK_BLOCKING_HINT}` : ""
290 }`;
291 yield* annotateTarget(target);
292
293 const connectivity = yield* Connectivity.Connectivity;
294 const driver = yield* ConnectionDriver.ConnectionDriver;
295 const wakeups = yield* ConnectionWakeups.ConnectionWakeups;
296 // The primary environment served this client or runs beside it on loopback,
297 // so the device reporting offline (often wrongly) never parks it. The report
298 // still probes its live session, like any other network change.
299 const targetNetwork = (network: NetworkStatus): NetworkStatus =>
300 network === "offline" && target._tag === "PrimaryConnectionTarget" ? "unknown" : network;
301 const initialIntent: SupervisorIntent = {
302 desired: options?.initiallyDesired ?? false,
303 network: targetNetwork(yield* connectivity.status),
304 };
305 const intent = yield* Ref.make(initialIntent);
306 const signals = yield* Queue.unbounded<SupervisorSignal>();
307 const resetRetryState = yield* Ref.make(false);
308 const probeUnanswered = yield* Ref.make(false);
309 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(
310 !initialIntent.desired
311 ? availableState(initialIntent, 0)
312 : initialIntent.network === "offline"
313 ? offlineState(initialIntent, 0, 0, null)
314 : connectingState(initialIntent, 0, 1, null),
315 );
316 const session = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(Option.none());
317 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
318 // Learned routes arrive while a session is live, so the route list is read
319 // fresh by each attempt and check instead of being fixed at creation.
320 const currentEntry = yield* Ref.make(entry);
321 const currentRoutes = Ref.get(currentEntry).pipe(Effect.map(connectionRoutes));
322 const preferredRouteId = yield* Ref.make(Option.none<string>());
323 const routeCooldowns = yield* Ref.make<ReadonlyMap<string, bigint>>(new Map());
324 // The generation keeps a check that started before a reset from recording
325 // its stale failures after it.
326 const silentRoutes = yield* Ref.make({
327 generation: 0,
328 routes: new Map<string, SilentRoute>() as ReadonlyMap<string, SilentRoute>,
329 });
330 const forgetSilentRoutes = Ref.update(silentRoutes, ({ generation }) => ({
331 generation: generation + 1,
332 routes: new Map(),
333 }));
334
335 const attemptEntry = Effect.gen(function* () {
336 const latest = yield* Ref.get(currentEntry);
337 const routes = connectionRoutes(latest);
338 const preferred = yield* Ref.getAndSet(preferredRouteId, Option.none());
339 if (Option.isNone(preferred)) return latest;
340 const route = routes.find(
341 (candidate) => connectionRouteId(candidate.target) === preferred.value,
342 );
343 return route === undefined
344 ? latest
345 : entryWithRoutes(latest, [route, ...routes.filter((candidate) => candidate !== route)]);
346 });
347
348 const learnRoutesFrom = Effect.fnUntraced(function* (
349 learn: NonNullable<EnvironmentSupervisorOptions["learnRoutes"]>,
350 lease: ConnectedLease,
351 ) {
352 const activeRoute = lease.activeRoute;
353 if (activeRoute === undefined) return;
354 yield* lease.session.subscribeServerConfig().pipe(
355 Stream.filter((event) => event.type === "snapshot"),
356 Stream.map((event) => event.config.directEndpoints),
357 Stream.filter((reported) => reported !== undefined),
358 Stream.changesWith((previous, next) => JSON.stringify(previous) === JSON.stringify(next)),
359 Stream.runForEach((reported) =>
360 Effect.gen(function* () {
361 const updated = yield* learn({ activeRoute, reported });
362 if (Option.isSome(updated)) {
363 yield* Ref.set(currentEntry, updated.value);
364 yield* requestBetterRouteCheck(lease);
365 }
366 }),
367 ),
368 Effect.ignoreCause({ log: true }),
369 );
370 });
371
372 const routeIndex = (lease: ConnectedLease) =>
373 currentRoutes.pipe(
374 Effect.map((routes) => {
375 const activeId = connectionRouteId(lease.prepared.target);
376 const index = routes.findIndex((route) => connectionRouteId(route.target) === activeId);
377 return index === -1 && lease.activeRoute !== undefined
378 ? insertRoute(routes, lease.activeRoute).findIndex(
379 (route) => connectionRouteId(route.target) === activeId,
380 )
381 : index;
382 }),
383 );
384
385 const checkBetterRoutes = Effect.fnUntraced(function* (lease: ConnectedLease) {
386 const latest = yield* Ref.get(currentEntry);
387 const current = yield* routeIndex(lease);
388 if (current <= 0) return;
389 const now = yield* Clock.monotonicTimeNanos;
390 const cooldowns = yield* Ref.get(routeCooldowns);
391 const silent = yield* Ref.get(silentRoutes);
392 const better = connectionRoutes(latest)
393 .slice(0, current)
394 .filter((route) => {
395 const id = connectionRouteId(route.target);
396 return (cooldowns.get(id) ?? 0n) <= now && (silent.routes.get(id)?.until ?? 0n) <= now;
397 });
398 const results = yield* Effect.forEach(
399 better,
400 (route: ConnectionRoute) => driver.preflight(latest, route),
401 { concurrency: "unbounded" },
402 );
403 yield* Ref.update(silentRoutes, (previous) => {
404 if (previous.generation !== silent.generation) return previous;
405 const routes = new Map(previous.routes);
406 for (const [position, route] of better.entries()) {
407 const id = connectionRouteId(route.target);
408 if (results[position] !== "silent") routes.delete(id);
409 else routes.set(id, backOffSilentRoute(previous.routes.get(id), now));
410 }
411 return { generation: previous.generation, routes };
412 });
413 const index = results.indexOf("ready");
414 if (index === -1) return;
415 const live = yield* SubscriptionRef.get(session);
416 if (Option.isNone(live) || live.value !== lease.session) return;
417 yield* signal({
418 _tag: "BetterRouteAvailable",
419 routeId: connectionRouteId(better[index]!.target),
420 });
421 });
422
423 const betterRouteChecks = yield* Queue.dropping<ConnectedLease>(1);
424 const requestBetterRouteCheck = (lease: ConnectedLease) =>
425 routeIndex(lease).pipe(
426 Effect.flatMap((index) =>
427 index > 0 ? Queue.offer(betterRouteChecks, lease).pipe(Effect.asVoid) : Effect.void,
428 ),
429 );
430
431 const clearLease = Effect.all(
432 [SubscriptionRef.set(session, Option.none()), SubscriptionRef.set(prepared, Option.none())],
433 { discard: true },
434 );
435
436 const setState = Effect.fn("EnvironmentSupervisor.setState")(function* (
437 next: SupervisorConnectionState,
438 ) {
439 yield* SubscriptionRef.set(state, next);
440 });
441
442 const signal = Effect.fn("EnvironmentSupervisor.signal")(function* (next: SupervisorSignal) {
443 yield* Queue.offer(signals, next);
444 });
445
446 const logManagedRelayAccountChange = Effect.logInfo(
447 "Managed relay account changed; restarting the environment connection.",
448 ).pipe(
449 Effect.annotateLogs({
450 "environment.id": target.environmentId,
451 "environment.label": target.label,
452 }),
453 );
454
455 const reportProgress = Effect.fn("EnvironmentSupervisor.reportProgress")(function* (
456 attempt: number,
457 generation: number,
458 lastFailure: ConnectionAttemptError | null,
459 progress: ConnectionDriver.ConnectionDriverProgress,
460 ) {
461 if ("prepared" in progress) {
462 yield* SubscriptionRef.set(prepared, Option.some(progress.prepared));
463 }
464 yield* setState(
465 connectingState(yield* Ref.get(intent), generation, attempt, lastFailure, progress.stage),
466 );
467 });
468
469 const establishConnection = Effect.fnUntraced(function* (
470 attempt: number,
471 generation: number,
472 lastFailure: ConnectionAttemptError | null,
473 synchronizedRoutes: Ref.Ref<number>,
474 ) {
475 return yield* driver.connect(yield* attemptEntry, (progress) =>
476 (progress.stage === "synchronizing"
477 ? Ref.update(synchronizedRoutes, (count) => count + 1)
478 : Effect.void
479 ).pipe(Effect.andThen(reportProgress(attempt, generation, lastFailure, progress))),
480 );
481 });
482
483 // Low-cardinality dimensions for time-to-connected dashboards; the error
484 // detail stays in the span status, never in an attribute.
485 const annotateAttemptOutcome = Effect.fnUntraced(function* (
486 exit: Exit.Exit<ConnectionDriver.EnvironmentConnectionLease, ConnectionAttemptError>,
487 ) {
488 if (Exit.isSuccess(exit)) {
489 const leaseTarget = exit.value.prepared.target;
490 const route = (yield* currentRoutes).find(
491 (candidate) => connectionRouteId(candidate.target) === connectionRouteId(leaseTarget),
492 );
493 return yield* Effect.annotateCurrentSpan({
494 "connection.outcome": "connected",
495 "connection.route": connectionRouteKind(
496 route ?? { target: leaseTarget, profile: Option.none() },
497 ),
498 });
499 }
500 if (Cause.hasInterruptsOnly(exit.cause)) {
501 return yield* Effect.annotateCurrentSpan("connection.outcome", "interrupted");
502 }
503 const failure = exit.cause.reasons.find(Cause.isFailReason)?.error;
504 yield* Effect.annotateCurrentSpan({
505 "connection.outcome": "failed",
506 "connection.failure.tag": failure?._tag ?? "defect",
507 "connection.failure.reason": failure?.reason ?? "defect",
508 });
509 });
510
511 const traceRelayEstablishment = (
512 effect: Effect.Effect<
513 ConnectionDriver.EnvironmentConnectionLease,
514 ConnectionAttemptError,
515 Scope.Scope
516 >,
517 attempt: number,
518 generation: number,
519 pendingRetry: Option.Option<PendingRetryTrace>,
520 ) => {
521 // The span must end with the ConnectionAttemptError itself: the exporter
522 // can only describe an Error-like failure, and a TracedAttemptFailure
523 // wrapper reached it as "[object Object]". The wrapper is added outside.
524 let attemptSpan = Option.none<Tracer.Span>();
525 const traced = Effect.gen(function* () {
526 attemptSpan = Option.some(yield* Effect.currentSpan.pipe(Effect.orDie));
527 yield* annotateTarget(target);
528 yield* Effect.annotateCurrentSpan({
529 "connection.attempt": attempt,
530 "connection.generation": generation,
531 "connection.retry.failure_count": Option.match(pendingRetry, {
532 onNone: () => 0,
533 onSome: (retry) => retry.failureCount,
534 }),
535 });
536 return yield* effect.pipe(Effect.onExit(annotateAttemptOutcome));
537 }).pipe(Effect.withSpan("relay.connection.attempt", { root: true }));
538
539 return Option.match(pendingRetry, {
540 onNone: () => traced,
541 onSome: (retry) =>
542 traced.pipe(
543 Effect.linkSpans(retry.previousAttempt, {
544 "connection.retry.delay_ms": retry.delayMs,
545 "connection.retry.reason": retry.reason,
546 }),
547 ),
548 }).pipe(
549 Effect.map((lease) => ({ attemptSpan, lease })),
550 Effect.mapError((error): TracedAttemptFailure => ({ error, attemptSpan })),
551 withRelayClientTracing,
552 );
553 };
554
555 const establishTracedConnection = Effect.fnUntraced(function* (
556 attempt: number,
557 generation: number,
558 lastFailure: ConnectionAttemptError | null,
559 pendingRetry: Option.Option<PendingRetryTrace>,
560 synchronizedRoutes: Ref.Ref<number>,
561 ) {
562 if (usesRelay) {
563 return yield* traceRelayEstablishment(
564 establishConnection(attempt, generation, lastFailure, synchronizedRoutes),
565 attempt,
566 generation,
567 pendingRetry,
568 );
569 }
570 return yield* establishConnection(attempt, generation, lastFailure, synchronizedRoutes).pipe(
571 Effect.map((lease) => ({
572 attemptSpan: Option.none<Tracer.Span>(),
573 lease,
574 })),
575 Effect.mapError((error): TracedAttemptFailure => ({
576 error,
577 attemptSpan: Option.none(),
578 })),
579 );
580 });
581
582 const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () {
583 // A primary attempt keeps running through an offline report, so the
584 // network returning restarts it in case it stalled meanwhile.
585 let deviceOffline = (yield* connectivity.status) === "offline";
586 for (;;) {
587 const next = yield* Queue.take(signals);
588 switch (next._tag) {
589 case "DisconnectRequested":
590 case "RetryRequested":
591 return false;
592 case "NetworkChanged":
593 if (target._tag !== "PrimaryConnectionTarget") {
594 if (next.network === "offline") {
595 return false;
596 }
597 } else if (next.network === "offline") {
598 deviceOffline = true;
599 } else if (deviceOffline) {
600 return true;
601 }
602 break;
603 case "ConnectRequested":
604 case "BetterRouteAvailable":
605 break;
606 case "Wakeup":
607 // An attempt that ran into a long suspension is likely stuck.
608 if (next.reason === "application-active-long") {
609 return true;
610 }
611 if (next.reason === "credentials-changed" && usesRelay) {
612 yield* logManagedRelayAccountChange;
613 return false;
614 }
615 break;
616 }
617 }
618 });
619
620 const isRelayLease = (lease: ConnectionDriver.EnvironmentConnectionLease) =>
621 lease.prepared.target._tag === "RelayConnectionTarget";
622
623 const connectedLeaseEnd = Effect.fnUntraced(function* (
624 next: SupervisorSignal,
625 lease: ConnectionDriver.EnvironmentConnectionLease,
626 ) {
627 if (next._tag === "DisconnectRequested") {
628 return "end" as const;
629 }
630 if (next._tag === "BetterRouteAvailable") {
631 yield* Ref.set(preferredRouteId, Option.some(next.routeId));
632 return "reset" as const;
633 }
634 // Only a session over T3 Connect holds the old account's credential.
635 if (next._tag === "Wakeup" && next.reason === "credentials-changed" && isRelayLease(lease)) {
636 yield* logManagedRelayAccountChange;
637 return "end" as const;
638 }
639 return undefined;
640 });
641
642 const leaseRouteKind = (lease: ConnectionDriver.EnvironmentConnectionLease) =>
643 currentRoutes.pipe(
644 Effect.map((routes) => {
645 const route = routes.find(
646 (candidate) =>
647 connectionRouteId(candidate.target) === connectionRouteId(lease.prepared.target),
648 );
649 return route === undefined ? undefined : connectionRouteKind(route);
650 }),
651 );
652
653 const isLongResume = (signal: SupervisorSignal) =>
654 signal._tag === "Wakeup" && signal.reason === "application-active-long";
655
656 const answersQuickly = (routeKind: ConnectionRouteKind | undefined) =>
657 routeKind === "loopback" || routeKind === "lan";
658
659 const probeTimeoutFor = (
660 next: SupervisorSignal,
661 routeKind: ConnectionRouteKind | undefined,
662 ): Duration.Input | undefined => {
663 switch (next._tag) {
664 case "RetryRequested":
665 return QUICK_CONNECTION_PROBE_TIMEOUT;
666 case "NetworkChanged":
667 return next.network === "offline" ? QUICK_CONNECTION_PROBE_TIMEOUT : undefined;
668 case "Wakeup":
669 if (next.reason === "application-active") {
670 return CONNECTION_PROBE_TIMEOUT;
671 }
672 // After a long suspension the data on screen is stale, so a slow
673 // answer reconnects as quickly as a missing one, even on a remote
674 // route where a reconnect takes longer than the wait it replaces.
675 if (next.reason === "application-active-long") {
676 return LONG_RESUME_PROBE_TIMEOUT;
677 }
678 if (next.reason !== "application-active-probe" && next.reason !== "network-changed") {
679 return undefined;
680 }
681 // Only loopback and LAN answer within 3 seconds reliably. A tailnet
682 // over DERP, a VPN, or the relay can spike past it on a live socket;
683 // a dead one is still caught early by the silent socket check.
684 return answersQuickly(routeKind)
685 ? QUICK_CONNECTION_PROBE_TIMEOUT
686 : CONNECTION_PROBE_TIMEOUT;
687 case "ConnectRequested":
688 case "DisconnectRequested":
689 case "BetterRouteAvailable":
690 return undefined;
691 }
692 };
693
694 const isForegroundProbe = (next: SupervisorSignal) =>
695 next._tag === "Wakeup" && next.reason === "application-active";
696
697 // Ends "silent" when the host answers a fresh request while the socket
698 // delivers nothing, as when iOS kills a suspended socket without a close
699 // event, and "alive" once the socket delivers a frame. A slow or
700 // unreachable host never ends it.
701 const awaitSilentSocket = Effect.fn("EnvironmentSupervisor.awaitSilentSocket")(function* (
702 lease: ConnectedLease,
703 routeKind: ConnectionRouteKind | undefined,
704 longResume: boolean,
705 ) {
706 const startedAt = yield* Clock.monotonicTimeNanos;
707 yield* Effect.sleep(SILENT_SOCKET_CHECK_AFTER);
708 const configLoaded = yield* lease.session.initialConfig.pipe(
709 Effect.as(true),
710 Effect.catchCause(() => Effect.succeed(false)),
711 );
712 if (!configLoaded) return yield* Effect.never;
713 // Frames that queued while the app was suspended have drained by now, so
714 // a later frame shows the path still works.
715 const framesBefore = yield* lease.session.inboundFrames;
716 const socketSilent = lease.session.inboundFrames.pipe(
717 Effect.map((frames) => frames === framesBefore),
718 );
719 // The session only counts frames, so a host check slower than the probe
720 // deadline would otherwise hide a frame until after the socket is replaced.
721 const frameArrives = Effect.sleep(FRAME_CHECK_EVERY).pipe(
722 Effect.andThen(socketSilent),
723 Effect.repeat({ while: (silent) => silent }),
724 Effect.as("alive" as const),
725 );
726 const hostAnswersWhileSilent = Effect.gen(function* () {
727 // A radio still waking up or a stale pooled connection can fail one check.
728 while ((yield* driver.checkHost(lease.prepared)) !== "answered") {
729 yield* Effect.sleep(SILENT_SOCKET_RECHECK_EVERY);
730 }
731 if (!answersQuickly(routeKind) && !longResume) {
732 const silentFor = (yield* Clock.monotonicTimeNanos) - startedAt;
733 const remaining = Duration.toNanosUnsafe(REMOTE_SILENT_SOCKET_AFTER) - silentFor;
734 if (remaining > 0n) yield* Effect.sleep(Duration.nanos(remaining));
735 }
736 return (yield* socketSilent) ? ("silent" as const) : ("alive" as const);
737 });
738 return yield* Effect.race(hostAnswersWhileSilent, frameArrives);
739 });
740
741 // Returning to the app, an explicit retry, and the network reporting offline
742 // all probe the live session instead of replacing it, so a healthy socket is
743 // not torn down (the offline report is often wrong, for example for a
744 // loopback server).
745 const monitorConnectedLease = Effect.fnUntraced(function* (lease: ConnectedLease) {
746 // A probe answers an explicit retry here, so the retry must not also reset
747 // the backoff of a later, unrelated failure.
748 const takeSignal = Queue.take(signals).pipe(
749 Effect.tap((next) =>
750 next._tag === "RetryRequested" ? Ref.set(resetRetryState, false) : Effect.void,
751 ),
752 );
753 // Ticks always run: the check skips the preferred route itself, and the
754 // route list can grow while connected (learned routes).
755 yield* Stream.tick(Duration.millis(BETTER_ROUTE_CHECK_INTERVAL_MS)).pipe(
756 Stream.drop(1),
757 Stream.runForEach(() => requestBetterRouteCheck(lease)),
758 Effect.forkScoped,
759 );
760 const routeKind = yield* leaseRouteKind(lease);
761 for (;;) {
762 const next = yield* takeSignal;
763 const end = yield* connectedLeaseEnd(next, lease);
764 if (end !== undefined) {
765 return end === "reset";
766 }
767 if (
768 (next._tag === "NetworkChanged" && next.network !== "offline") ||
769 (next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason))
770 ) {
771 yield* requestBetterRouteCheck(lease);
772 }
773 const probeTimeout = probeTimeoutFor(next, routeKind);
774 if (probeTimeout === undefined) {
775 continue;
776 }
777 yield* Ref.set(probeUnanswered, true);
778 let probe = yield* Effect.forkChild(lease.session.probe);
779 // A busy server can miss one foreground probe while the socket is still
780 // healthy, so the first miss asks again and only the follow-up's miss
781 // replaces the session.
782 let toleratesMiss = isForegroundProbe(next);
783 let silentSocket: Fiber.Fiber<"silent" | "alive"> | undefined;
784 // A signal during a running check, such as a network change that kills
785 // the socket, needs a fresh check only if the running one sees a frame.
786 let recheck = false;
787 let longResume = isLongResume(next);
788 const startSilentSocketCheck = Effect.gen(function* () {
789 silentSocket = yield* Effect.forkChild(awaitSilentSocket(lease, routeKind, longResume));
790 recheck = false;
791 });
792 // The web and desktop foreground probe allows a busy host 15 seconds,
793 // and a busy host can answer a fresh request before a queued probe.
794 const armSilentSocketCheck = Effect.fnUntraced(function* (signal: SupervisorSignal) {
795 if (isForegroundProbe(signal)) return;
796 if (silentSocket !== undefined) {
797 recheck = true;
798 return;
799 }
800 yield* startSilentSocketCheck;
801 });
802 yield* armSilentSocketCheck(next);
803 const stopProbe = Effect.suspend(() =>
804 Effect.all(
805 [Fiber.interrupt(probe), silentSocket ? Fiber.interrupt(silentSocket) : Effect.void],
806 { discard: true },
807 ),
808 );
809 let longResumeStartedAt = yield* Clock.monotonicTimeNanos;
810 let deadline = longResumeStartedAt + Duration.toNanosUnsafe(probeTimeout);
811 for (;;) {
812 const remaining = deadline - (yield* Clock.monotonicTimeNanos);
813 const probeEvent = yield* Effect.raceAllFirst([
814 Fiber.await(probe).pipe(
815 Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })),
816 ),
817 takeSignal.pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))),
818 Effect.sleep(Duration.nanos(remaining > 0n ? remaining : 0n)).pipe(
819 Effect.as({ _tag: "TimedOut" as const }),
820 ),
821 (silentSocket ? Fiber.join(silentSocket) : Effect.never).pipe(
822 Effect.map((check) =>
823 check === "silent"
824 ? { _tag: "SocketSilent" as const }
825 : { _tag: "SocketAlive" as const },
826 ),
827 ),
828 ]);
829 if (probeEvent._tag === "SocketAlive") {
830 silentSocket = undefined;
831 // A frame shows the socket still works, so its probe gets the usual resume time.
832 if (longResume) {
833 const aliveDeadline =
834 longResumeStartedAt + Duration.toNanosUnsafe(QUICK_CONNECTION_PROBE_TIMEOUT);
835 if (aliveDeadline > deadline) deadline = aliveDeadline;
836 }
837 if (recheck) yield* startSilentSocketCheck;
838 continue;
839 }
840 if (probeEvent._tag === "SocketSilent") {
841 // The host is reachable, so replace the socket at once, even while
842 // the network reports offline.
843 yield* stopProbe;
844 return true;
845 }
846 if (probeEvent._tag === "TimedOut") {
847 if (toleratesMiss) {
848 toleratesMiss = false;
849 yield* Fiber.interrupt(probe);
850 probe = yield* Effect.forkChild(lease.session.probe);
851 deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout);
852 continue;
853 }
854 yield* stopProbe;
855 // A long suspension leaves the socket suspect even when the network
856 // reports offline, so it is replaced like a silent one.
857 if (longResume) return true;
858 return yield* new ConnectionTransientError({
859 reason: "timeout",
860 detail: `${target.label} did not respond to a connection health check.`,
861 });
862 }
863 if (probeEvent._tag === "ProbeCompleted") {
864 if (silentSocket) yield* Fiber.interrupt(silentSocket);
865 if (Exit.isSuccess(probeEvent.exit)) {
866 yield* Ref.set(probeUnanswered, false);
867 }
868 yield* probeEvent.exit;
869 break;
870 }
871 const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease);
872 if (endDuringProbe !== undefined) {
873 yield* stopProbe;
874 return endDuringProbe === "reset";
875 }
876 const signalTimeout = probeTimeoutFor(probeEvent.signal, routeKind);
877 if (signalTimeout !== undefined) {
878 // A retry or network signal wants a definite answer, even when its
879 // own deadline lands after the foreground one.
880 if (!isForegroundProbe(probeEvent.signal)) toleratesMiss = false;
881 if (isLongResume(probeEvent.signal) && !longResume) {
882 longResume = true;
883 longResumeStartedAt = yield* Clock.monotonicTimeNanos;
884 }
885 yield* armSilentSocketCheck(probeEvent.signal);
886 const signalDeadline =
887 (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout);
888 if (signalDeadline < deadline) deadline = signalDeadline;
889 }
890 }
891 }
892 });
893
894 const runAttempt = Effect.fnUntraced(function* (
895 attempt: number,
896 generation: number,
897 lastFailure: ConnectionAttemptError | null,
898 pendingRetry: Option.Option<PendingRetryTrace>,
899 ignoreOffline: boolean,
900 ) {
901 const switchingTo = yield* Ref.get(preferredRouteId);
902 yield* SubscriptionRef.set(prepared, Option.none());
903 const synchronizedRoutes = yield* Ref.make(0);
904 const establishment = yield* Effect.raceAllFirst([
905 exitUnlessInterrupted(
906 establishTracedConnection(
907 attempt,
908 generation,
909 lastFailure,
910 pendingRetry,
911 synchronizedRoutes,
912 ),
913 ).pipe(
914 Effect.map((exit): EstablishmentEvent => ({
915 _tag: "Completed",
916 exit,
917 })),
918 ),
919 waitForEstablishmentInterrupt().pipe(
920 Effect.map((resetRetry): EstablishmentEvent => ({
921 _tag: "Interrupted",
922 resetRetry,
923 })),
924 ),
925 // Each route may use the full setup time before the next is tried, and
926 // each route that reaches synchronizing adds its own snapshot budget, so
927 // a route that fails late does not eat the next route's time.
928 Effect.gen(function* () {
929 const routes = connectionRoutes(yield* Ref.get(currentEntry));
930 const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routes.length));
931 // The desktop bounds SSH preparation, which may download a server
932 // release, so for SSH-only entries the clock stops while preparing.
933 const pausesWhilePreparing = routes.every((route) => connectionRouteKind(route) === "ssh");
934 const preparing = (expected: boolean) =>
935 SubscriptionRef.changes(state).pipe(
936 Stream.filter((current) => (current.stage === "preparing") === expected),
937 Stream.runHead,
938 );
939 let spent = 0n;
940 while (true) {
941 if (pausesWhilePreparing) yield* preparing(false);
942 const synchronized = BigInt(yield* Ref.get(synchronizedRoutes));
943 const remaining =
944 setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout) - spent;
945 if (remaining <= 0n) {
946 return { _tag: "TimedOut" } as const satisfies EstablishmentEvent;
947 }
948 const resumedAt = yield* Clock.monotonicTimeNanos;
949 const sleep = Effect.sleep(Duration.nanos(remaining));
950 yield* pausesWhilePreparing ? Effect.raceFirst(sleep, preparing(true)) : sleep;
951 spent += (yield* Clock.monotonicTimeNanos) - resumedAt;
952 }
953 }),
954 ]);
955
956 if (establishment._tag === "Interrupted") {
957 return {
958 _tag: "Interrupted",
959 established: false,
960 stable: false,
961 resetRetry: establishment.resetRetry,
962 } satisfies AttemptOutcome;
963 }
964 if (establishment._tag === "TimedOut") {
965 return {
966 _tag: "Failure",
967 established: false,
968 stable: false,
969 failure: {
970 error: new ConnectionTransientError({
971 reason: "timeout",
972 detail: setupTimeoutDetail,
973 }),
974 attemptSpan: Option.none(),
975 },
976 } satisfies AttemptOutcome;
977 }
978 if (Exit.isFailure(establishment.exit)) {
979 const isUnexpectedDefect =
980 !Cause.hasInterruptsOnly(establishment.exit.cause) &&
981 !establishment.exit.cause.reasons.some(Cause.isFailReason);
982 const outcome = failureFromExit(target, establishment.exit, false, false);
983 if (isUnexpectedDefect) {
984 const defect = establishment.exit.cause.reasons.find(Cause.isDieReason)?.defect;
985 yield* Effect.logError("Connection attempt failed with an unexpected defect.").pipe(
986 Effect.annotateLogs({
987 "environment.id": target.environmentId,
988 "environment.label": target.label,
989 "cause.reason_count": establishment.exit.cause.reasons.length,
990 ...safeErrorLogAttributes(defect),
991 }),
992 );
993 }
994 return outcome;
995 }
996
997 const active = establishment.exit.value;
998 if (Option.isSome(switchingTo)) {
999 const landed = connectionRouteId(active.lease.prepared.target);
1000 if (landed !== switchingTo.value) {
1001 const until =
1002 (yield* Clock.monotonicTimeNanos) + BigInt(BETTER_ROUTE_COOLDOWN_MS) * 1_000_000n;
1003 yield* Ref.update(routeCooldowns, (current) =>
1004 new Map(current).set(switchingTo.value, until),
1005 );
1006 }
1007 }
1008 const currentIntent = yield* Ref.get(intent);
1009 if (!currentIntent.desired || (currentIntent.network === "offline" && !ignoreOffline)) {
1010 return {
1011 _tag: "Interrupted",
1012 established: false,
1013 stable: false,
1014 resetRetry: false,
1015 } satisfies AttemptOutcome;
1016 }
1017
1018 const connectedLease: ConnectedLease = {
1019 ...active.lease,
1020 activeRoute: (yield* currentRoutes).find(
1021 (route) =>
1022 connectionRouteId(route.target) === connectionRouteId(active.lease.prepared.target),
1023 ),
1024 };
1025 const connectedAt = yield* Clock.currentTimeMillis;
1026 yield* SubscriptionRef.set(prepared, Option.some(connectedLease.prepared));
1027 yield* SubscriptionRef.set(session, Option.some(connectedLease.session));
1028 if (options?.learnRoutes !== undefined) {
1029 yield* learnRoutesFrom(options.learnRoutes, connectedLease).pipe(Effect.forkScoped);
1030 }
1031 yield* setState({
1032 desired: true,
1033 network: currentIntent.network,
1034 phase: "connected",
1035 stage: null,
1036 attempt,
1037 generation,
1038 lastFailure: null,
1039 retryAt: null,
1040 });
1041
1042 const connectedExit = yield* Effect.raceFirst(
1043 connectedLease.session.closed.pipe(
1044 Effect.mapError((error): TracedAttemptFailure => ({
1045 error,
1046 attemptSpan: active.attemptSpan,
1047 })),
1048 ),
1049 monitorConnectedLease(connectedLease).pipe(
1050 Effect.mapError((error): TracedAttemptFailure => ({
1051 error,
1052 attemptSpan: active.attemptSpan,
1053 })),
1054 ),
1055 ).pipe(exitUnlessInterrupted);
1056 const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt;
1057 if (Exit.isSuccess(connectedExit)) {
1058 return {
1059 _tag: "Interrupted",
1060 established: true,
1061 stable: connectedForMs >= BACKOFF_RESET_AFTER_MS,
1062 resetRetry: connectedExit.value,
1063 } satisfies AttemptOutcome;
1064 }
1065 const outcome = failureFromExit(
1066 target,
1067 connectedExit,
1068 true,
1069 connectedForMs >= BACKOFF_RESET_AFTER_MS,
1070 );
1071 if (outcome._tag === "Failure") {
1072 yield* Effect.void.pipe(
1073 Effect.withSpan("EnvironmentSupervisor.connectionLost", {
1074 root: true,
1075 attributes: {
1076 "environment.id": target.environmentId,
1077 "environment.label": target.label,
1078 "environment.target.kind": target._tag,
1079 "connection.connected_ms": connectedForMs,
1080 "connection.failure.reason": outcome.failure.error.reason,
1081 "connection.failure.detail": outcome.failure.error.detail,
1082 },
1083 }),
1084 );
1085 }
1086 return outcome;
1087 }, Effect.ensuring(clearLease));
1088
1089 const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) {
1090 // @effect-diagnostics-next-line raceFirstWithSleepToTimeout:off - the sleep is the retry delay (false), not a timeout around the signal loop
1091 return yield* Effect.raceFirst(
1092 Effect.sleep(delayMs).pipe(Effect.as(false)),
1093 Effect.gen(function* () {
1094 for (;;) {
1095 const next = yield* Queue.take(signals);
1096 switch (next._tag) {
1097 case "Wakeup":
1098 return ConnectionWakeups.resetsRetryBackoff(next.reason);
1099 case "ConnectRequested":
1100 case "DisconnectRequested":
1101 case "RetryRequested":
1102 case "NetworkChanged":
1103 return false;
1104 case "BetterRouteAvailable":
1105 break;
1106 }
1107 }
1108 }),
1109 );
1110 });
1111
1112 const waitForSignal = Queue.take(signals).pipe(
1113 Effect.repeat({ while: (next) => next._tag === "BetterRouteAvailable" }),
1114 Effect.map(
1115 (next) => next._tag === "Wakeup" && ConnectionWakeups.resetsRetryBackoff(next.reason),
1116 ),
1117 );
1118
1119 const run = Effect.fnUntraced(function* () {
1120 let failureCount = 0;
1121 let generation = 0;
1122 let latestFailure: ConnectionAttemptError | null = null;
1123 let pendingRetry = Option.none<PendingRetryTrace>();
1124 const resetRetryLadder = () => {
1125 failureCount = 0;
1126 pendingRetry = Option.none();
1127 };
1128 let replacing = false;
1129
1130 for (;;) {
1131 if (yield* Ref.getAndSet(resetRetryState, false)) {
1132 failureCount = 0;
1133 latestFailure = null;
1134 pendingRetry = Option.none();
1135 }
1136 const currentIntent = yield* Ref.get(intent);
1137 if (!currentIntent.desired) {
1138 resetRetryLadder();
1139 latestFailure = null;
1140 yield* clearLease;
1141 yield* setState(availableState(currentIntent, generation));
1142 yield* waitForSignal;
1143 continue;
1144 }
1145 if (currentIntent.network === "offline" && !replacing) {
1146 yield* clearLease;
1147 yield* setState(offlineState(currentIntent, generation, failureCount + 1, latestFailure));
1148 const applicationActivated = yield* waitForSignal;
1149 if (applicationActivated) {
1150 resetRetryLadder();
1151 }
1152 continue;
1153 }
1154
1155 const attempt = failureCount + 1;
1156 const nextGeneration = generation + 1;
1157 const outcome: AttemptOutcome = yield* Effect.scoped(
1158 runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, replacing),
1159 );
1160 replacing = false;
1161 const failedProbe = yield* Ref.getAndSet(probeUnanswered, false);
1162 if (outcome.established) {
1163 generation = nextGeneration;
1164 // A session came up, so no earlier failure describes the connection.
1165 // Losing a live session is not a failed connection either, for example
1166 // a socket the OS killed while the app was suspended: its replacement
1167 // reports no failure until it fails itself.
1168 latestFailure = null;
1169 if (outcome.stable) {
1170 resetRetryLadder();
1171 }
1172 }
1173 if (outcome._tag === "Interrupted") {
1174 if (outcome.resetRetry) {
1175 resetRetryLadder();
1176 replacing = true;
1177 }
1178 continue;
1179 }
1180
1181 const attemptSpan: Option.Option<Tracer.Span> = outcome.failure.attemptSpan;
1182 const error: ConnectionAttemptError = outcome.failure.error;
1183 if (!outcome.established) {
1184 latestFailure = error;
1185 }
1186 if (error._tag === "ConnectionBlockedError") {
1187 const blockedIntent = yield* Ref.get(intent);
1188 yield* setState({
1189 desired: blockedIntent.desired,
1190 network: blockedIntent.network,
1191 phase: "blocked",
1192 stage: null,
1193 attempt,
1194 generation,
1195 lastFailure: error,
1196 retryAt: null,
1197 });
1198 const applicationActivated = yield* waitForSignal;
1199 if (applicationActivated) {
1200 resetRetryLadder();
1201 }
1202 continue;
1203 }
1204
1205 if (failedProbe) {
1206 resetRetryLadder();
1207 yield* setState(connectingState(yield* Ref.get(intent), generation, 1, latestFailure));
1208 continue;
1209 }
1210
1211 // The first loss of a live session retries almost at once; the ladder
1212 // still counts it, so a host that keeps dropping sessions backs off.
1213 const firstLoss = outcome.established && failureCount === 0;
1214 failureCount += 1;
1215 const delayMs = firstLoss
1216 ? Math.round(SESSION_LOSS_RETRY_DELAY_MS * (0.5 + (yield* Random.next) / 2))
1217 : retryDelayMs(failureCount - 1, yield* Random.next);
1218 pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({
1219 previousAttempt,
1220 failureCount,
1221 delayMs,
1222 reason: error.reason,
1223 }));
1224 const failedIntent = yield* Ref.get(intent);
1225 yield* setState({
1226 desired: failedIntent.desired,
1227 network: failedIntent.network,
1228 phase: "backoff",
1229 stage: null,
1230 attempt,
1231 generation,
1232 lastFailure: latestFailure,
1233 retryAt: (yield* Clock.currentTimeMillis) + delayMs,
1234 });
1235 const applicationActivated = yield* waitForRetrySignal(delayMs);
1236 if (applicationActivated) {
1237 resetRetryLadder();
1238 }
1239 }
1240 });
1241
1242 yield* connectivity.changes.pipe(
1243 Stream.runForEach((network) =>
1244 Ref.modify(intent, (current) => {
1245 const next = targetNetwork(network);
1246 return current.network === next
1247 ? [false, current]
1248 : ([true, { ...current, network: next }] as const);
1249 }).pipe(
1250 Effect.flatMap((changed) =>
1251 changed
1252 ? (network === "offline" ? Effect.void : forgetSilentRoutes).pipe(
1253 Effect.andThen(signal({ _tag: "NetworkChanged", network })),
1254 )
1255 : Effect.void,
1256 ),
1257 ),
1258 ),
1259 Effect.forkScoped,
1260 );
1261 yield* wakeups.changes.pipe(
1262 Stream.runForEach((reason) =>
1263 // A long suspension can hide a network change the platform never reported.
1264 (reason === "network-changed" || reason === "application-active-long"
1265 ? forgetSilentRoutes
1266 : Effect.void
1267 ).pipe(Effect.andThen(signal({ _tag: "Wakeup", reason }))),
1268 ),
1269 Effect.forkScoped,
1270 );
1271 yield* Queue.take(betterRouteChecks).pipe(
1272 Effect.flatMap(checkBetterRoutes),
1273 Effect.forever,
1274 Effect.forkScoped,
1275 );
1276 yield* run().pipe(Effect.forkScoped);
1277
1278 const connect = Ref.update(intent, (current) => ({
1279 ...current,
1280 desired: true,
1281 })).pipe(
1282 Effect.andThen(signal({ _tag: "ConnectRequested" })),
1283 Effect.withSpan("EnvironmentSupervisor.connect"),
1284 );
1285
1286 const disconnect = Ref.update(intent, (current) => ({
1287 ...current,
1288 desired: false,
1289 })).pipe(
1290 Effect.andThen(signal({ _tag: "DisconnectRequested" })),
1291 Effect.withSpan("EnvironmentSupervisor.disconnect"),
1292 );
1293
1294 const retryNow = Ref.set(resetRetryState, true).pipe(
1295 Effect.andThen(signal({ _tag: "RetryRequested" })),
1296 Effect.withSpan("EnvironmentSupervisor.retryNow"),
1297 );
1298
1299 yield* Effect.addFinalizer(() => Queue.shutdown(signals).pipe(Effect.andThen(clearLease)));
1300
1301 return EnvironmentSupervisor.of({
1302 target,
1303 state,
1304 session,
1305 prepared,
1306 connect,
1307 disconnect,
1308 retryNow,
1309 });
1310});