Open History to see the last commit
1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Exit from "effect/Exit";
5import * as Fiber from "effect/Fiber";
6import * as Layer from "effect/Layer";
7import * as Scope from "effect/Scope";
8import * as FetchHttpClient from "effect/http/FetchHttpClient";
9import * as HttpClient from "effect/http/HttpClient";
10
11import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts";
12import type {
13 ConnectionAttemptError,
14 ConnectionAttemptStage,
15 PreparedConnection,
16} from "./model.ts";
17import { ConnectionTransientError } from "./model.ts";
18import * as ConnectionResolver from "./resolver.ts";
19import {
20 connectionRoutes,
21 routeEntry,
22 routeHttpBaseUrl,
23 routeKeepsItsHost,
24 sendsTokenInClear,
25 usableRoutes,
26} from "./routes.ts";
27import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
28import * as RpcSession from "../rpc/session.ts";
29import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts";
30
31export type ConnectionDriverProgress =
32 | {
33 readonly stage: "preparing";
34 }
35 | {
36 readonly stage: Exclude<ConnectionAttemptStage, "preparing">;
37 readonly prepared: PreparedConnection;
38 };
39
40export interface EnvironmentConnectionLease {
41 readonly prepared: PreparedConnection;
42 readonly session: RpcSession.RpcSession;
43}
44
45export type RouteCheck = "answered" | "silent" | "unchecked";
46/** "unprepared" means the route answered its check but could not be prepared. */
47export type RoutePreflight = "ready" | "unprepared" | "silent" | "unchecked";
48
49const ROUTE_CHECK_TIMEOUT_MS = 2_500;
50// Preferred direct routes share one head start, so a relay connection does not
51// wait the full probe timeout for a LAN the device has already left. A route
52// that misses it connects over relay, then upgrades with a second reconnect.
53const ROUTE_CHECK_HEAD_START_MS = 500;
54
55export class ConnectionDriver extends Context.Service<
56 ConnectionDriver,
57 {
58 readonly connect: (
59 entry: ConnectionCatalogEntry,
60 reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect<void>,
61 ) => Effect.Effect<EnvironmentConnectionLease, ConnectionAttemptError, Scope.Scope>;
62 readonly preflight: (
63 entry: ConnectionCatalogEntry,
64 route: ConnectionRoute,
65 ) => Effect.Effect<RoutePreflight>;
66 /** Asks the host behind a live session to answer a fresh request on the session's own origin. */
67 readonly checkHost: (prepared: PreparedConnection) => Effect.Effect<RouteCheck>;
68 }
69>()("@t3tools/client-runtime/connection/driver/ConnectionDriver") {}
70
71export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes")(function* <R>(
72 entry: ConnectionCatalogEntry,
73 checkRoute: (route: ConnectionRoute) => Effect.Effect<RouteCheck>,
74 connectRoute: (
75 route: ConnectionRoute,
76 ) => Effect.Effect<EnvironmentConnectionLease, ConnectionAttemptError, R | Scope.Scope>,
77) {
78 const routes = usableRoutes(entry);
79 const checks =
80 routes.length === 1
81 ? []
82 : yield* Effect.forEach(routes, (route) => Effect.forkChild(checkRoute(route)));
83 const headStart = yield* Effect.forkChild(Effect.sleep(ROUTE_CHECK_HEAD_START_MS));
84 const attemptScope = yield* Scope.Scope;
85 let transient: ConnectionAttemptError | undefined;
86 let blocked: ConnectionAttemptError | undefined;
87 // Each route gets its own scope so a half-open session closes before the next try.
88 const attempt = Effect.fnUntraced(function* (route: ConnectionRoute) {
89 const routeScope = yield* Scope.fork(attemptScope);
90 const result = yield* connectRoute(route).pipe(
91 Scope.provide(routeScope),
92 Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : Scope.close(routeScope, exit))),
93 Effect.result,
94 );
95 if (result._tag === "Failure") {
96 if (result.failure._tag === "ConnectionTransientError") transient ??= result.failure;
97 else blocked ??= result.failure;
98 }
99 return result;
100 });
101 const silent: Array<{ readonly route: ConnectionRoute; readonly index: number }> = [];
102 for (const [index, route] of routes.entries()) {
103 const check =
104 checks.length === 0
105 ? "unchecked"
106 : yield* Effect.raceFirst(
107 Fiber.join(checks[index]!),
108 Fiber.join(headStart).pipe(Effect.as<RouteCheck>("silent")),
109 );
110 if (check === "silent") {
111 silent.push({ route, index });
112 continue;
113 }
114 const result = yield* attempt(route);
115 if (result._tag === "Success") return result.success;
116 // An incompatible server is the same server on every route.
117 if (result.failure.reason === "unsupported") return yield* result.failure;
118 }
119 const late = yield* Effect.forEach(silent, ({ index }) => Fiber.join(checks[index]!), {
120 concurrency: "unbounded",
121 });
122 const answeredLate = silent.filter((_, position) => late[position] !== "silent");
123 const stillSilent = silent.filter((_, position) => late[position] === "silent");
124 for (const { route } of [...answeredLate, ...stillSilent]) {
125 // A route still silent after its whole check can hold the attempt open far
126 // longer than the check did. Once another route failed in a way that
127 // retries, the retry reaches it sooner than waiting here would.
128 if (transient !== undefined && stillSilent.some((entry) => entry.route === route)) break;
129 const result = yield* attempt(route);
130 if (result._tag === "Success") return result.success;
131 if (result.failure.reason === "unsupported") return yield* result.failure;
132 }
133 return yield* (
134 transient ??
135 blocked ??
136 new ConnectionTransientError({
137 reason: "endpoint-unavailable",
138 detail: `${entry.target.label} did not answer on any saved route.`,
139 })
140 );
141});
142
143/** @public */
144export const make = Effect.gen(function* () {
145 const resolver = yield* ConnectionResolver.ConnectionResolver;
146 const sessions = yield* RpcSession.RpcSessionFactory;
147 const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
148 const httpClient = yield* HttpClient.HttpClient;
149
150 // The descriptor is public, so this sends no credential to whatever
151 // answers at a saved LAN address on a different network.
152 const checkDescriptor = (httpBaseUrl: string, environmentId: string, keepsItsHost: boolean) =>
153 fetchRemoteEnvironmentDescriptor({
154 httpBaseUrl,
155 timeoutMs: ROUTE_CHECK_TIMEOUT_MS,
156 }).pipe(
157 Effect.option,
158 // The connect this check leads to skips its own read. Every answer, even a
159 // failure, replaces what an older check remembered.
160 Effect.tap((descriptor) =>
161 remote.rememberCheckedDescriptor(httpBaseUrl, descriptor, keepsItsHost),
162 ),
163 Effect.map((descriptor): RouteCheck =>
164 Option.exists(descriptor, (value) => value.environmentId === environmentId)
165 ? "answered"
166 : "silent",
167 ),
168 // A cached descriptor would answer for a host that cannot be reached.
169 Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }),
170 Effect.provideService(HttpClient.HttpClient, httpClient),
171 );
172
173 const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => {
174 const httpBaseUrl = routeHttpBaseUrl(route);
175 if (httpBaseUrl === null) return Effect.succeed<RouteCheck>("unchecked");
176 return checkDescriptor(httpBaseUrl, entry.target.environmentId, routeKeepsItsHost(route)).pipe(
177 Effect.withSpan("ConnectionDriver.checkRoute", {
178 attributes: { "connection.target.kind": route.target._tag },
179 }),
180 );
181 };
182
183 const checkHost = (prepared: PreparedConnection) =>
184 checkDescriptor(
185 prepared.httpBaseUrl,
186 prepared.environmentId,
187 prepared.httpBaseUrl.startsWith("https:"),
188 ).pipe(
189 Effect.withSpan("ConnectionDriver.checkHost", {
190 attributes: { "connection.target.kind": prepared.target._tag },
191 }),
192 );
193
194 const connect = Effect.fn("ConnectionDriver.connect")(function* (
195 entry: ConnectionCatalogEntry,
196 reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect<void>,
197 ) {
198 const target = entry.target;
199 yield* Effect.annotateCurrentSpan({
200 "connection.environment.id": target.environmentId,
201 "connection.target.kind": target._tag,
202 "connection.route.count": connectionRoutes(entry).length,
203 });
204 yield* reportProgress({ stage: "preparing" });
205 return yield* connectOverRoutes(
206 entry,
207 (route) => checkRoute(entry, route),
208 Effect.fnUntraced(function* (route) {
209 yield* reportProgress({ stage: "preparing" });
210 const prepared = yield* resolver.prepare(routeEntry(entry, route));
211 yield* reportProgress({ stage: "opening", prepared });
212 const session = yield* sessions.connect(prepared);
213 yield* reportProgress({ stage: "synchronizing", prepared });
214 yield* session.ready;
215 // The server closes a socket when its session expires, unless the same key renewed it.
216 if (prepared.httpAuthorization?._tag === "Dpop") {
217 yield* remote
218 .renewDpopBeforeExpiry({
219 expectedEnvironmentId: prepared.environmentId,
220 expiresAtEpochMs: prepared.httpAuthorization.expiresAtEpochMs,
221 })
222 .pipe(
223 Effect.catch((error) =>
224 Effect.logWarning("Could not renew the connection's credential.", { error }),
225 ),
226 Effect.forkScoped,
227 );
228 }
229 return { prepared, session } satisfies EnvironmentConnectionLease;
230 }),
231 );
232 });
233
234 const preflight = (entry: ConnectionCatalogEntry, route: ConnectionRoute) =>
235 (sendsTokenInClear(connectionRoutes(entry), route)
236 ? Effect.succeed("silent" as const)
237 : checkRoute(entry, route)
238 ).pipe(
239 Effect.filterOrElse(
240 (check): check is Exclude<RouteCheck, "answered"> => check !== "answered",
241 () =>
242 resolver.prepare(routeEntry(entry, route)).pipe(
243 Effect.as<RoutePreflight>("ready"),
244 Effect.orElseSucceed((): RoutePreflight => "unprepared"),
245 ),
246 ),
247 Effect.withSpan("ConnectionDriver.preflight"),
248 );
249
250 return ConnectionDriver.of({ connect, preflight, checkHost });
251});
252
253export const layer = Layer.effect(ConnectionDriver, make);