1import {
2 ORCHESTRATION_PROTOCOL_VERSION,
3 type EnvironmentId,
4 type ExecutionEnvironmentDescriptor,
5 type ServerSelfUpdateInput,
6 type ServerSelfUpdateResult,
7 WS_METHODS,
8} from "@t3tools/contracts";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as Ref from "effect/Ref";
14import * as Schedule from "effect/Schedule";
15import * as Schema from "effect/Schema";
16import * as Stream from "effect/Stream";
17import * as SubscriptionRef from "effect/SubscriptionRef";
18import * as HttpClient from "effect/http/HttpClient";
19import * as RpcClient from "effect/rpc/RpcClient";
20import * as RpcSerialization from "effect/rpc/RpcSerialization";
21import * as Socket from "effect/socket/Socket";
22
23import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts";
24import { makeWsRpcProtocolClient, PING_TIMEOUT } from "../rpc/protocol.ts";
25import { isLegacyUpdateHandoffLoss, resolveServerUpdateProgressResult } from "../state/server.ts";
26import * as RelayEnvironmentDiscovery from "../relay/discovery.ts";
27import * as ConnectionResolver from "./resolver.ts";
28import * as EnvironmentRegistry from "./registry.ts";
29import { connectionRoutes, hasRelayRoute, routeEntry, routeHttpBaseUrl } from "./routes.ts";
30
31// A v1 host restarting into v2 runs migrations before its descriptor answers again.
32const OUTDATED_HOST_RESTART_TIMEOUT = Duration.minutes(4);
33const SOCKET_OPEN_TIMEOUT = "15 seconds";
34
35export class OutdatedHostUpdateError extends Schema.TaggedError<OutdatedHostUpdateError>()(
36 "OutdatedHostUpdateError",
37 {
38 environmentId: Schema.String,
39 message: Schema.String,
40 },
41) {}
42
43export type OutdatedHostUpdateStage = "downloading" | "installing" | "resuming";
44
45/**
46 * Updates a host whose orchestration protocol is too old for this client.
47 *
48 * The normal session refuses to open against such a host, so this opens a
49 * bare socket and calls only the self-update RPCs, whose wire shape has not
50 * changed across protocol versions. Once the host relaunches on a compatible
51 * protocol the environment is switched back on and connects normally.
52 */
53export const updateOutdatedHost = Effect.fn("clientRuntime.connection.updateOutdatedHost")(
54 function* (
55 environmentId: EnvironmentId,
56 input: ServerSelfUpdateInput,
57 onStage: (stage: OutdatedHostUpdateStage) => Effect.Effect<void>,
58 ) {
59 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
60 const resolver = yield* ConnectionResolver.ConnectionResolver;
61 const webSocketConstructor = yield* Socket.WebSocketConstructor;
62 const httpClient = yield* HttpClient.HttpClient;
63 const entry = (yield* SubscriptionRef.get(registry.entries)).get(environmentId);
64 if (entry === undefined) {
65 return yield* new EnvironmentRegistry.EnvironmentNotRegisteredError({ environmentId });
66 }
67 // An outdated server cannot connect normally, so the routes are tried in
68 // order here. The first that authorizes carries the update.
69 const { prepared, descriptor } = yield* Effect.firstSuccessOf(
70 connectionRoutes(entry).map((route) => resolver.prepareForUpdate(routeEntry(entry, route))),
71 );
72 const capabilities = descriptor.capabilities;
73 if (
74 capabilities.serverSelfUpdate === undefined ||
75 (capabilities.serverSelfUpdate === "desktop-managed" &&
76 capabilities.desktopAppUpdate !== true)
77 ) {
78 return yield* new OutdatedHostUpdateError({
79 environmentId,
80 message: `Update the server on ${descriptor.label} manually; it cannot update itself.`,
81 });
82 }
83
84 const result = yield* Effect.scoped(
85 Effect.gen(function* () {
86 const protocolContext = yield* Layer.build(
87 Layer.effect(
88 RpcClient.Protocol,
89 RpcClient.makeProtocolSocket({
90 pingTimeout: PING_TIMEOUT,
91 retryTransientErrors: false,
92 retryPolicy: Schedule.recurs(0),
93 }),
94 ).pipe(
95 Layer.provide(
96 Layer.mergeAll(
97 Socket.layerWebSocket(prepared.socketUrl, {
98 openTimeout: SOCKET_OPEN_TIMEOUT,
99 }).pipe(
100 Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)),
101 ),
102 RpcSerialization.layerJson,
103 ),
104 ),
105 ),
106 );
107 const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
108
109 const updateResult: ServerSelfUpdateResult =
110 capabilities.serverSelfUpdateProgress === true
111 ? yield* Effect.gen(function* () {
112 const terminal = yield* Ref.make(Option.none<ServerSelfUpdateResult>());
113 const streamExit = yield* client[WS_METHODS.serverUpdateServerWithProgress](
114 input,
115 ).pipe(
116 Stream.runForEach((event) =>
117 event.type === "complete"
118 ? Ref.set(terminal, Option.some(event.result))
119 : onStage(event.stage),
120 ),
121 Effect.exit,
122 );
123 return yield* resolveServerUpdateProgressResult(
124 input.targetVersion,
125 yield* Ref.get(terminal),
126 streamExit,
127 );
128 })
129 : yield* client[WS_METHODS.serverUpdateServer](input).pipe(
130 // Older servers can drop the socket before acknowledging the restart.
131 Effect.catchCauseIf(
132 (cause) =>
133 (capabilities.serverSelfUpdate === "boot-service" ||
134 capabilities.serverSelfUpdate === "respawn") &&
135 isLegacyUpdateHandoffLoss(cause),
136 () =>
137 Effect.succeed({
138 targetVersion: input.targetVersion,
139 method: capabilities.serverSelfUpdate as "boot-service" | "respawn",
140 } satisfies ServerSelfUpdateResult),
141 ),
142 );
143
144 if (
145 updateResult.method === "desktop-app" &&
146 updateResult.desktopUpdateToken !== undefined
147 ) {
148 // The commit relaunches the desktop app, so a dropped socket is success.
149 yield* client[WS_METHODS.serverCommitDesktopUpdate]({
150 requestId: updateResult.desktopUpdateToken,
151 }).pipe(
152 Effect.catchCauseIf(
153 (cause) => isLegacyUpdateHandoffLoss(cause),
154 () => Effect.void,
155 ),
156 );
157 }
158 return updateResult;
159 }),
160 );
161
162 yield* onStage("resuming");
163 // The restarted host may answer on another saved route, so each poll asks
164 // every direct address, the one that carried the update first.
165 const pollUrls = [
166 prepared.httpBaseUrl,
167 ...connectionRoutes(entry).flatMap((route) => routeHttpBaseUrl(route) ?? []),
168 ].filter((url, index, all) => all.indexOf(url) === index);
169 const resumed = yield* Effect.firstSuccessOf(
170 pollUrls.map((httpBaseUrl) =>
171 fetchRemoteEnvironmentDescriptor({ httpBaseUrl }).pipe(
172 Effect.filterOrFail(
173 (descriptor) =>
174 descriptor.environmentId === environmentId && isCompatibleDescriptor(descriptor),
175 () => "not-ready" as const,
176 ),
177 ),
178 ),
179 ).pipe(
180 Effect.provideService(HttpClient.HttpClient, httpClient),
181 Effect.option,
182 Effect.repeat({
183 schedule: Schedule.spaced(Duration.seconds(1)),
184 until: (current) => Option.exists(current, isCompatibleDescriptor),
185 }),
186 Effect.timeoutOption(OUTDATED_HOST_RESTART_TIMEOUT),
187 Effect.map(Option.flatten),
188 );
189 if (Option.isNone(resumed)) {
190 return yield* new OutdatedHostUpdateError({
191 environmentId,
192 message: `${descriptor.label} did not come back on a compatible server version.`,
193 });
194 }
195
196 // Discovery still holds the old relay descriptor and would re-block the
197 // environment from it, so replace that before clearing the block.
198 if (hasRelayRoute(entry)) {
199 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
200 yield* discovery.refresh;
201 }
202 yield* registry.setCompatibility(environmentId, null);
203 yield* registry.setEnabled(environmentId, true);
204 return { ...result, targetVersion: resumed.value.serverVersion };
205 },
206);
207
208function isCompatibleDescriptor(descriptor: ExecutionEnvironmentDescriptor): boolean {
209 return (descriptor.orchestrationProtocolVersion ?? 1) === ORCHESTRATION_PROTOCOL_VERSION;
210}