Open History to see the last commit
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 {
30 connectionRoutes,
31 hasRelayRoute,
32 routeEntry,
33 routeHttpBaseUrl,
34 usableRoutes,
35} from "./routes.ts";
36
37// A v1 host restarting into v2 runs migrations before its descriptor answers again.
38const OUTDATED_HOST_RESTART_TIMEOUT = Duration.minutes(4);
39const SOCKET_OPEN_TIMEOUT = "15 seconds";
40
41export class OutdatedHostUpdateError extends Schema.TaggedError<OutdatedHostUpdateError>()(
42 "OutdatedHostUpdateError",
43 {
44 environmentId: Schema.String,
45 message: Schema.String,
46 },
47) {}
48
49export type OutdatedHostUpdateStage = "downloading" | "installing" | "resuming";
50
51const reachOutdatedHost = Effect.fn("clientRuntime.connection.reachOutdatedHost")(function* (
52 environmentId: EnvironmentId,
53) {
54 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
55 const resolver = yield* ConnectionResolver.ConnectionResolver;
56 const entry = (yield* SubscriptionRef.get(registry.entries)).get(environmentId);
57 if (entry === undefined) {
58 return yield* new EnvironmentRegistry.EnvironmentNotRegisteredError({ environmentId });
59 }
60 const { prepared, descriptor } = yield* Effect.firstSuccessOf(
61 usableRoutes(entry).map((route) => resolver.prepareForUpdate(routeEntry(entry, route))),
62 );
63 return { entry, prepared, descriptor };
64});
65
66// The normal session cannot read config from a host behind the protocol gate.
67export const describeOutdatedHost = (environmentId: EnvironmentId) =>
68 reachOutdatedHost(environmentId).pipe(Effect.map(({ descriptor }) => descriptor));
69
70/**
71 * The normal session refuses to open against such a host, so this opens a bare
72 * socket and calls only the self-update RPCs, whose wire shape has not changed
73 * across protocol versions.
74 */
75export const updateOutdatedHost = Effect.fn("clientRuntime.connection.updateOutdatedHost")(
76 function* (
77 environmentId: EnvironmentId,
78 input: ServerSelfUpdateInput,
79 onStage: (stage: OutdatedHostUpdateStage) => Effect.Effect<void>,
80 ) {
81 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
82 const webSocketConstructor = yield* Socket.WebSocketConstructor;
83 const httpClient = yield* HttpClient.HttpClient;
84 const { entry, prepared, descriptor } = yield* reachOutdatedHost(environmentId);
85 // Discovery still holds the old relay descriptor and would re-block the
86 // environment from it, so replace that before clearing the block.
87 const switchBackOn = Effect.gen(function* () {
88 if (hasRelayRoute(entry)) {
89 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
90 yield* discovery.refresh;
91 }
92 yield* registry.setCompatibility(environmentId, null);
93 yield* registry.setEnabled(environmentId, true);
94 });
95
96 // A host updated by hand is already compatible, and a bare socket would be
97 // rejected with 426, so there is nothing to update.
98 if (isCompatibleDescriptor(descriptor)) {
99 yield* switchBackOn;
100 return { targetVersion: descriptor.serverVersion };
101 }
102
103 const capabilities = descriptor.capabilities;
104 if (
105 capabilities.serverSelfUpdate === undefined ||
106 (capabilities.serverSelfUpdate === "desktop-managed" &&
107 capabilities.desktopAppUpdate !== true)
108 ) {
109 return yield* new OutdatedHostUpdateError({
110 environmentId,
111 message: `Update T3 Code on ${descriptor.label} manually; it cannot update itself.`,
112 });
113 }
114
115 const result = yield* Effect.scoped(
116 Effect.gen(function* () {
117 const protocolContext = yield* Layer.build(
118 Layer.effect(
119 RpcClient.Protocol,
120 RpcClient.makeProtocolSocket({
121 pingTimeout: PING_TIMEOUT,
122 retryTransientErrors: false,
123 retryPolicy: Schedule.recurs(0),
124 }),
125 ).pipe(
126 Layer.provide(
127 Layer.mergeAll(
128 Socket.layerWebSocket(prepared.socketUrl, {
129 openTimeout: SOCKET_OPEN_TIMEOUT,
130 }).pipe(
131 Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)),
132 ),
133 RpcSerialization.layerJson,
134 ),
135 ),
136 ),
137 );
138 const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
139
140 const updateResult: ServerSelfUpdateResult =
141 capabilities.serverSelfUpdateProgress === true
142 ? yield* Effect.gen(function* () {
143 const terminal = yield* Ref.make(Option.none<ServerSelfUpdateResult>());
144 let receivedProgress = false;
145 const streamExit = yield* client[WS_METHODS.serverUpdateServerWithProgress](
146 input,
147 ).pipe(
148 Stream.runForEach((event) =>
149 event.type === "complete"
150 ? Ref.set(terminal, Option.some(event.result))
151 : Effect.sync(() => {
152 receivedProgress = true;
153 }).pipe(Effect.andThen(onStage(event.stage))),
154 ),
155 Effect.exit,
156 );
157 const selfUpdateMethod = capabilities.serverSelfUpdate;
158 // Desktop preparation needs its terminal commit token.
159 const accepted =
160 receivedProgress &&
161 (selfUpdateMethod === "boot-service" || selfUpdateMethod === "respawn")
162 ? { targetVersion: input.targetVersion, method: selfUpdateMethod }
163 : undefined;
164 return yield* resolveServerUpdateProgressResult(
165 input.targetVersion,
166 yield* Ref.get(terminal),
167 streamExit,
168 accepted,
169 );
170 })
171 : yield* client[WS_METHODS.serverUpdateServer](input).pipe(
172 // Older servers can drop the socket before acknowledging the restart.
173 Effect.catchCauseIf(
174 (cause) =>
175 (capabilities.serverSelfUpdate === "boot-service" ||
176 capabilities.serverSelfUpdate === "respawn") &&
177 isLegacyUpdateHandoffLoss(cause),
178 () =>
179 Effect.succeed({
180 targetVersion: input.targetVersion,
181 method: capabilities.serverSelfUpdate as "boot-service" | "respawn",
182 } satisfies ServerSelfUpdateResult),
183 ),
184 );
185
186 if (
187 updateResult.method === "desktop-app" &&
188 updateResult.desktopUpdateToken !== undefined
189 ) {
190 // The commit relaunches the desktop app, so a dropped socket is success.
191 yield* client[WS_METHODS.serverCommitDesktopUpdate]({
192 requestId: updateResult.desktopUpdateToken,
193 }).pipe(
194 Effect.catchCauseIf(
195 (cause) => isLegacyUpdateHandoffLoss(cause),
196 () => Effect.void,
197 ),
198 );
199 }
200 return updateResult;
201 }),
202 );
203
204 yield* onStage("resuming");
205 const pollUrls = [
206 prepared.httpBaseUrl,
207 ...connectionRoutes(entry).flatMap((route) => routeHttpBaseUrl(route) ?? []),
208 ].filter((url, index, all) => all.indexOf(url) === index);
209 const resumed = yield* Effect.firstSuccessOf(
210 pollUrls.map((httpBaseUrl) =>
211 fetchRemoteEnvironmentDescriptor({ httpBaseUrl }).pipe(
212 Effect.filterOrFail(
213 (descriptor) =>
214 descriptor.environmentId === environmentId && isCompatibleDescriptor(descriptor),
215 () => "not-ready" as const,
216 ),
217 ),
218 ),
219 ).pipe(
220 Effect.provideService(HttpClient.HttpClient, httpClient),
221 Effect.option,
222 Effect.repeat({
223 schedule: Schedule.spaced(Duration.seconds(1)),
224 until: (current) => Option.exists(current, isCompatibleDescriptor),
225 }),
226 Effect.timeoutOption(OUTDATED_HOST_RESTART_TIMEOUT),
227 Effect.map(Option.flatten),
228 );
229 if (Option.isNone(resumed)) {
230 return yield* new OutdatedHostUpdateError({
231 environmentId,
232 message: `${descriptor.label} did not come back on a compatible T3 Code version.`,
233 });
234 }
235
236 yield* switchBackOn;
237 return { ...result, targetVersion: resumed.value.serverVersion };
238 },
239);
240
241function isCompatibleDescriptor(descriptor: ExecutionEnvironmentDescriptor): boolean {
242 return (descriptor.orchestrationProtocolVersion ?? 1) === ORCHESTRATION_PROTOCOL_VERSION;
243}