diff --git a/apps/web/src/components/ServerUpdateAction.tsx b/apps/web/src/components/ServerUpdateAction.tsx index fa7ec5e601f196a419115c46ee40a4d7835a8df8..82307156ba5884bab60a23c1230206c85e40bd35 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -10,7 +10,7 @@ import { type ComponentProps, useRef, useState } from "react"; import { requestConfirmDialog } from "~/confirmDialog"; import { useCopyToClipboard } from "~/hooks/useCopyToClipboard"; import { useEnvironmentSettings } from "~/hooks/useSettings"; -import { serverEnvironment } from "~/state/server"; +import { serverEnvironment, updateOutdatedServer } from "~/state/server"; import { useAtomCommand } from "~/state/use-atom-command"; import { manualServerUpdateCommand } from "~/versionSkew"; import { Button } from "./ui/button"; @@ -290,3 +290,56 @@ export function ServerUpdateAction({ ); } + +/** + * Updates a host too old for this client to connect to. Its version comes + * from the host descriptor because the host never delivers a server config. + */ +export function OutdatedServerUpdateAction({ + environmentId, + serverLabel, + fromVersion, + targetVersion, + label = "Update", +}: { + readonly environmentId: EnvironmentId; + readonly serverLabel: string; + readonly fromVersion: string | undefined; + readonly targetVersion: string; + readonly label?: string; +}) { + const update = useAtomCommand(updateOutdatedServer, { reportFailure: false }); + const handleUpdate = async () => { + if (pendingUpdateEnvironmentIds.has(environmentId)) return; + pendingUpdateEnvironmentIds.add(environmentId); + try { + const result = await update({ + environmentId, + input: { targetVersion }, + ...(fromVersion === undefined ? {} : { fromVersion }), + }); + if (result._tag === "Failure") { + if (isAtomCommandInterrupted(result)) return; + throw squashAtomCommandFailure(result); + } + toastManager.add({ + type: "success", + title: `${serverLabel} updated`, + description: `Reconnected on t3@${result.value.targetVersion}.`, + }); + } catch (error) { + toastManager.add({ + type: "error", + title: "Server update failed", + description: updateFailureMessage(error), + }); + } finally { + pendingUpdateEnvironmentIds.delete(environmentId); + } + }; + return ( + + ); +} diff --git a/apps/web/src/components/settings/ConnectionsSettings.tsx b/apps/web/src/components/settings/ConnectionsSettings.tsx index 7804bfbdfa2cc2727553407ea68c4033946bd129..e018e7941ec17ab9d9dc381b31908d7837cf1672 100644 --- a/apps/web/src/components/settings/ConnectionsSettings.tsx +++ b/apps/web/src/components/settings/ConnectionsSettings.tsx @@ -165,11 +165,13 @@ import { usePrimaryEnvironment, useRelayEnvironmentDiscovery, } from "~/state/environments"; +import { APP_VERSION } from "~/branding"; import { requestConfirmDialog } from "~/confirmDialog"; import { useAtomCommand } from "../../state/use-atom-command"; import { primaryServerKeybindingsAtom, serverEnvironment } from "~/state/server"; import { ConnectionStatusDot } from "../ConnectionStatusDot"; import { + OutdatedServerUpdateAction, ServerUpdateAction, ServerUpdateProgress, ServerUpdatesAction, @@ -1598,6 +1600,17 @@ function SavedBackendListRow({ ) : null } > + {unsupported && + environment.entry.serverUpdateRequired === true && + serverUpdateState.status !== "running" ? ( + + ) : null} {showUpdateAction ? ( ()( diff --git a/packages/client-runtime/src/connection/compatibility.test.ts b/packages/client-runtime/src/connection/compatibility.test.ts index e6a2cf3ca18f618e2570023bc48426ac624b823b..3f9dc8f99245428be2f2d372bba701550e7ffb77 100644 --- a/packages/client-runtime/src/connection/compatibility.test.ts +++ b/packages/client-runtime/src/connection/compatibility.test.ts @@ -50,5 +50,29 @@ describe("orchestration protocol compatibility", () => { ); expect(error).toMatchObject({ reason: "unsupported" }); expect(error?.message).toContain("This client is not supported"); + expect(error).not.toHaveProperty("serverUpdateRequired"); + }); + + it("offers a remote update only for an older host that can update itself", () => { + const older = descriptor(ORCHESTRATION_PROTOCOL_VERSION - 1); + const withCapabilities = (capabilities: ExecutionEnvironmentDescriptor["capabilities"]) => + orchestrationProtocolCompatibilityError({ ...older, capabilities }); + + expect( + withCapabilities({ repositoryIdentity: true, serverSelfUpdate: "boot-service" }), + ).toMatchObject({ serverUpdateRequired: true }); + expect(withCapabilities({ repositoryIdentity: true })).not.toHaveProperty( + "serverUpdateRequired", + ); + expect( + withCapabilities({ repositoryIdentity: true, serverSelfUpdate: "desktop-managed" }), + ).not.toHaveProperty("serverUpdateRequired"); + expect( + withCapabilities({ + repositoryIdentity: true, + serverSelfUpdate: "desktop-managed", + desktopAppUpdate: true, + }), + ).toMatchObject({ serverUpdateRequired: true }); }); }); diff --git a/packages/client-runtime/src/connection/compatibility.ts b/packages/client-runtime/src/connection/compatibility.ts index 0a08ddb2acfabfcf16e8e5c84334e6fa316fa0e5..c0ec739e55e0dd408320dec87501a179d6bc86c6 100644 --- a/packages/client-runtime/src/connection/compatibility.ts +++ b/packages/client-runtime/src/connection/compatibility.ts @@ -14,13 +14,25 @@ export function orchestrationProtocolCompatibilityError( if (serverProtocolVersion === ORCHESTRATION_PROTOCOL_VERSION) { return null; } - return new ConnectionBlockedError({ - reason: "unsupported", - detail: - serverProtocolVersion > ORCHESTRATION_PROTOCOL_VERSION - ? `This client is not supported by this server. Update your app or use a compatible release to connect to ${descriptor.label}.` - : `This client requires a newer server. Update T3 Code on ${descriptor.label} to connect.`, - }); + return serverProtocolVersion > ORCHESTRATION_PROTOCOL_VERSION + ? new ConnectionBlockedError({ + reason: "unsupported", + detail: `This client is not supported by this server. Update your app or use a compatible release to connect to ${descriptor.label}.`, + }) + : new ConnectionBlockedError({ + reason: "unsupported", + detail: `This client requires a newer server. Update T3 Code on ${descriptor.label} to connect.`, + ...(canSelfUpdate(descriptor) ? { serverUpdateRequired: true } : {}), + }); +} + +/** Whether this client can drive the host's update remotely. */ +function canSelfUpdate(descriptor: ExecutionEnvironmentDescriptor): boolean { + const { serverSelfUpdate, desktopAppUpdate } = descriptor.capabilities; + return ( + serverSelfUpdate !== undefined && + (serverSelfUpdate !== "desktop-managed" || desktopAppUpdate === true) + ); } export function appendOrchestrationProtocol(socketUrl: string): string { diff --git a/packages/client-runtime/src/connection/index.ts b/packages/client-runtime/src/connection/index.ts index cd6312163681218088142862542f798ee78cc241..04833c3feb37552770a5276c61be015f0808df42 100644 --- a/packages/client-runtime/src/connection/index.ts +++ b/packages/client-runtime/src/connection/index.ts @@ -16,3 +16,5 @@ export * as EnvironmentSupervisor from "./supervisor.ts"; export * as Wakeups from "./wakeups.ts"; export { orchestrationProtocolCompatibilityError } from "./compatibility.ts"; +// Flat so consumers' inferred command types can name it. +export { OutdatedHostUpdateError } from "./outdatedHostUpdate.ts"; diff --git a/packages/client-runtime/src/connection/layer.ts b/packages/client-runtime/src/connection/layer.ts index 8b1a4eaa5ddeb72ff5de4fe25efdc2258599f135..8ec233fcc0974e66833d99da8894086e8f02cb6f 100644 --- a/packages/client-runtime/src/connection/layer.ts +++ b/packages/client-runtime/src/connection/layer.ts @@ -75,6 +75,8 @@ export function layerWithOptions(options: RpcSession.RpcSessionOptions) { registryLayer, RelayEnvironmentDiscovery.layer, onboardingLayer, + // Exposed for updating hosts too old to connect through the driver. + ConnectionResolver.layer, ); const connectionStartupLayer = Layer.effectDiscard( Effect.gen(function* () { diff --git a/packages/client-runtime/src/connection/model.ts b/packages/client-runtime/src/connection/model.ts index 45d1866ec713cc350b196c267e0bb4a45472653d..78d11b84b9e48fbd24f808453f1f649b5ffcd5fc 100644 --- a/packages/client-runtime/src/connection/model.ts +++ b/packages/client-runtime/src/connection/model.ts @@ -94,6 +94,8 @@ export class ConnectionBlockedError extends Schema.TaggedError, - options?: { readonly failDescriptor?: boolean; readonly protocolVersion?: number }, + options?: { + readonly failDescriptor?: boolean; + readonly protocolVersion?: number; + readonly selfUpdate?: boolean; + }, ) { const fetchFn = ((input, init = {}) => { const url = String(input); @@ -56,6 +60,7 @@ function pairingHttpLayer( orchestrationProtocolVersion: options?.protocolVersion ?? ORCHESTRATION_PROTOCOL_VERSION, capabilities: { repositoryIdentity: true, + ...(options?.selfUpdate === true ? { serverSelfUpdate: "boot-service" } : {}), }, }), ); @@ -145,6 +150,51 @@ describe("connection onboarding", () => { }), ); + it.effect("pairs an outdated server so it can be updated from this client", () => + Effect.gen(function* () { + const calls: Array<{ readonly url: string; readonly init: RequestInit }> = []; + const registration = yield* preparePairingRegistration({ + host: "remote.example.test", + pairingCode: "pairing-token", + }).pipe( + Effect.provide( + Layer.mergeAll( + CLIENT_PRESENTATION_LAYER, + pairingHttpLayer(calls, { + protocolVersion: ORCHESTRATION_PROTOCOL_VERSION - 1, + selfUpdate: true, + }), + ), + ), + ); + expect(registration.target.environmentId).toBe("environment-paired"); + expect(calls.map((call) => call.url)).toContain("https://remote.example.test/oauth/token"); + }), + ); + + it.effect("refuses an outdated server that cannot update itself", () => + Effect.gen(function* () { + const calls: Array<{ readonly url: string; readonly init: RequestInit }> = []; + const error = yield* preparePairingRegistration({ + host: "remote.example.test", + pairingCode: "pairing-token", + }).pipe( + Effect.provide( + Layer.mergeAll( + CLIENT_PRESENTATION_LAYER, + pairingHttpLayer(calls, { protocolVersion: ORCHESTRATION_PROTOCOL_VERSION - 1 }), + ), + ), + Effect.flip, + ); + expect(error).toMatchObject({ reason: "unsupported" }); + expect(error).not.toHaveProperty("serverUpdateRequired"); + expect(calls.map((call) => call.url)).toEqual([ + "https://remote.example.test/.well-known/t3/environment", + ]); + }), + ); + it.effect("does not consume a pairing credential when descriptor discovery fails", () => Effect.gen(function* () { const calls: Array<{ readonly url: string; readonly init: RequestInit }> = []; diff --git a/packages/client-runtime/src/connection/onboarding.ts b/packages/client-runtime/src/connection/onboarding.ts index 24c03addfa66b75b5c4003f6151e2e6b7218a386..fec7847b541f15c80fefa5e3eef025b8ceb4ca40 100644 --- a/packages/client-runtime/src/connection/onboarding.ts +++ b/packages/client-runtime/src/connection/onboarding.ts @@ -93,7 +93,10 @@ export const preparePairingRegistration = Effect.fn( httpBaseUrl: target.httpBaseUrl, }).pipe(Effect.mapError(mapRemoteEnvironmentError)); const compatibilityError = orchestrationProtocolCompatibilityError(descriptor); - if (compatibilityError !== null) return yield* compatibilityError; + // An outdated server is still saved so it can be updated from this client. + if (compatibilityError !== null && compatibilityError.serverUpdateRequired !== true) { + return yield* compatibilityError; + } const access = yield* bootstrapRemoteBearerSession({ httpBaseUrl: target.httpBaseUrl, credential: target.credential, diff --git a/packages/client-runtime/src/connection/outdatedHostUpdate.test.ts b/packages/client-runtime/src/connection/outdatedHostUpdate.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..e63d095facb51268763f8d3470ba8564e6b7ee39 --- /dev/null +++ b/packages/client-runtime/src/connection/outdatedHostUpdate.test.ts @@ -0,0 +1,278 @@ +import { + EnvironmentId, + ORCHESTRATION_PROTOCOL_VERSION, + ExecutionEnvironmentDescriptor, + WS_METHODS, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as HttpClient from "effect/unstable/http/HttpClient"; +import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"; +import * as Socket from "effect/unstable/socket/Socket"; + +import type { ConnectionCatalogEntry } from "./catalog.ts"; +import { orchestrationProtocolCompatibilityError } from "./compatibility.ts"; +import { PrimaryConnectionTarget } from "./model.ts"; +import { updateOutdatedHost } from "./outdatedHostUpdate.ts"; +import * as EnvironmentRegistry from "./registry.ts"; +import * as ConnectionResolver from "./resolver.ts"; +import * as RelayEnvironmentDiscovery from "../relay/discovery.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-old"), + label: "Build Mac", + httpBaseUrl: "https://build.example.test", + wsBaseUrl: "wss://build.example.test", +}); + +const descriptor = (protocol: number | undefined, serverVersion: string) => + ({ + environmentId: TARGET.environmentId, + label: TARGET.label, + platform: { os: "darwin", arch: "arm64" }, + serverVersion, + ...(protocol === undefined ? {} : { orchestrationProtocolVersion: protocol }), + capabilities: { repositoryIdentity: true, serverSelfUpdate: "boot-service" }, + }) satisfies ExecutionEnvironmentDescriptor; + +const RpcRequest = Schema.TaggedStruct("Request", { + id: Schema.Union([Schema.String, Schema.Number]), + payload: Schema.Unknown, + tag: Schema.String, +}); +const isRpcRequest = Schema.is(RpcRequest); +const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); +const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); +const encodeDescriptor = Schema.encodeSync(Schema.fromJsonString(ExecutionEnvironmentDescriptor)); + +type Listener = (event: { readonly type: string; readonly data?: unknown }) => void; + +/** Answers update RPCs the way a protocol-1 server does. */ +class OutdatedHostSocket { + static readonly OPEN = 1; + readyState = 0; + readonly requests: Array = []; + private readonly listeners = new Map>(); + + readonly url: string; + private readonly onUpdate: () => void; + + constructor(url: string, onUpdate: () => void) { + this.url = url; + this.onUpdate = onUpdate; + queueMicrotask(() => { + this.readyState = OutdatedHostSocket.OPEN; + this.emit({ type: "open" }); + }); + } + + addEventListener(type: string, listener: Listener) { + const listeners = this.listeners.get(type) ?? new Set(); + listeners.add(listener); + this.listeners.set(type, listeners); + } + + removeEventListener(type: string, listener: Listener) { + this.listeners.get(type)?.delete(listener); + } + + send(data: string) { + const message = decodeJson(data); + if (!isRpcRequest(message)) return; + this.requests.push(message); + if (message.tag !== WS_METHODS.serverUpdateServer) return; + this.onUpdate(); + queueMicrotask(() => + this.emit({ + type: "message", + data: encodeJson({ + _tag: "Exit", + requestId: message.id, + exit: { + _tag: "Success", + value: { targetVersion: "0.0.46", method: "boot-service", updateId: "update-1" }, + }, + }), + }), + ); + } + + close() { + this.readyState = 3; + this.emit({ type: "close" }); + } + + private emit(event: { readonly type: string; readonly data?: unknown }) { + for (const listener of this.listeners.get(event.type) ?? []) listener(event); + } +} + +describe("updateOutdatedHost", () => { + it.effect("updates a protocol-1 host over a bare socket, then switches it back on", () => + Effect.gen(function* () { + const blocked = orchestrationProtocolCompatibilityError(descriptor(undefined, "0.0.45")); + expect(blocked).toMatchObject({ serverUpdateRequired: true }); + + const sockets: Array = []; + let served: ExecutionEnvironmentDescriptor = descriptor(undefined, "0.0.45"); + const entries = yield* SubscriptionRef.make< + ReadonlyMap + >( + new Map([ + [ + TARGET.environmentId, + { + target: TARGET, + profile: Option.none(), + enabled: false, + unsupportedReason: blocked?.message ?? "", + serverUpdateRequired: true, + }, + ], + ]), + ); + const calls: Array = []; + const registry = EnvironmentRegistry.EnvironmentRegistry.of({ + entries, + setCompatibility: (_environmentId: EnvironmentId, error: unknown) => + Effect.sync(() => calls.push(`compatibility:${error === null ? "clear" : "block"}`)), + setEnabled: (_environmentId: EnvironmentId, enabled: boolean) => + Effect.sync(() => calls.push(`enabled:${enabled}`)), + } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const resolver = ConnectionResolver.ConnectionResolver.of({ + prepare: () => Effect.die(new Error("The update must bypass the protocol gate.")), + prepareForUpdate: () => + Effect.sync(() => served).pipe( + Effect.map((current) => ({ + descriptor: current, + prepared: { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: "wss://build.example.test/ws?wsTicket=ticket", + httpAuthorization: null, + target: TARGET, + }, + })), + ), + }); + const httpClient = HttpClient.make((request) => + Effect.sync(() => + HttpClientResponse.fromWeb(request, new Response(encodeDescriptor(served))), + ), + ); + + const result = yield* updateOutdatedHost( + TARGET.environmentId, + { targetVersion: "0.0.46" }, + () => Effect.void, + ).pipe( + Effect.provide( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, registry), + Layer.succeed(ConnectionResolver.ConnectionResolver, resolver), + Layer.succeed( + RelayEnvironmentDiscovery.RelayEnvironmentDiscovery, + RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.of({ + state: yield* SubscriptionRef.make( + RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE, + ), + refresh: Effect.void, + }), + ), + Layer.succeed(HttpClient.HttpClient, httpClient), + Layer.succeed(Socket.WebSocketConstructor, (url) => { + // The host relaunches on a compatible protocol once the update lands. + const socket = new OutdatedHostSocket(url, () => { + served = descriptor(ORCHESTRATION_PROTOCOL_VERSION, "0.0.46"); + }); + sockets.push(socket); + return socket as unknown as globalThis.WebSocket; + }), + ), + ), + ); + + expect(sockets[0]?.url).not.toContain("orchestrationProtocol"); + expect(sockets[0]?.requests.map((request) => request.tag)).toEqual([ + WS_METHODS.serverUpdateServer, + ]); + expect(result.targetVersion).toBe("0.0.46"); + expect(calls).toEqual(["compatibility:clear", "enabled:true"]); + }), + ); + + it.effect("refuses a host that cannot update itself without opening a socket", () => + Effect.gen(function* () { + const manual = { + ...descriptor(undefined, "0.0.45"), + capabilities: { repositoryIdentity: true }, + }; + let opened = false; + const error = yield* Effect.flip( + updateOutdatedHost(TARGET.environmentId, { targetVersion: "0.0.46" }, () => Effect.void), + ).pipe( + Effect.provide( + Layer.mergeAll( + Layer.succeed( + EnvironmentRegistry.EnvironmentRegistry, + EnvironmentRegistry.EnvironmentRegistry.of({ + entries: yield* SubscriptionRef.make< + ReadonlyMap + >( + new Map([ + [ + TARGET.environmentId, + { target: TARGET, profile: Option.none(), enabled: false }, + ], + ]), + ), + } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]), + ), + Layer.succeed( + ConnectionResolver.ConnectionResolver, + ConnectionResolver.ConnectionResolver.of({ + prepare: () => Effect.die(new Error("unused")), + prepareForUpdate: () => + Effect.succeed({ + descriptor: manual, + prepared: { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: "wss://build.example.test/ws", + httpAuthorization: null, + target: TARGET, + }, + }), + }), + ), + Layer.succeed( + RelayEnvironmentDiscovery.RelayEnvironmentDiscovery, + RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.of({ + state: yield* SubscriptionRef.make( + RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE, + ), + refresh: Effect.void, + }), + ), + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make(() => Effect.die(new Error("unused"))), + ), + Layer.succeed(Socket.WebSocketConstructor, () => { + opened = true; + throw new Error("unused"); + }), + ), + ), + ); + expect(error).toMatchObject({ _tag: "OutdatedHostUpdateError" }); + expect(opened).toBe(false); + }), + ); +}); diff --git a/packages/client-runtime/src/connection/outdatedHostUpdate.ts b/packages/client-runtime/src/connection/outdatedHostUpdate.ts new file mode 100644 index 0000000000000000000000000000000000000000..0ca445b893864bb0a343679c55a999a7d786da31 --- /dev/null +++ b/packages/client-runtime/src/connection/outdatedHostUpdate.ts @@ -0,0 +1,190 @@ +import { + ORCHESTRATION_PROTOCOL_VERSION, + type EnvironmentId, + type ExecutionEnvironmentDescriptor, + type ServerSelfUpdateInput, + type ServerSelfUpdateResult, + WS_METHODS, +} from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as HttpClient from "effect/unstable/http/HttpClient"; +import * as RpcClient from "effect/unstable/rpc/RpcClient"; +import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"; +import * as Socket from "effect/unstable/socket/Socket"; + +import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts"; +import { makeWsRpcProtocolClient } from "../rpc/protocol.ts"; +import { isLegacyUpdateHandoffLoss, resolveServerUpdateProgressResult } from "../state/server.ts"; +import * as RelayEnvironmentDiscovery from "../relay/discovery.ts"; +import * as ConnectionResolver from "./resolver.ts"; +import * as EnvironmentRegistry from "./registry.ts"; + +// A v1 host restarting into v2 runs migrations before its descriptor answers again. +const OUTDATED_HOST_RESTART_TIMEOUT = Duration.minutes(4); +const SOCKET_OPEN_TIMEOUT = "15 seconds"; + +export class OutdatedHostUpdateError extends Schema.TaggedError()( + "OutdatedHostUpdateError", + { + environmentId: Schema.String, + message: Schema.String, + }, +) {} + +export type OutdatedHostUpdateStage = "downloading" | "installing" | "resuming"; + +/** + * Updates a host whose orchestration protocol is too old for this client. + * + * The normal session refuses to open against such a host, so this opens a + * bare socket and calls only the self-update RPCs, whose wire shape has not + * changed across protocol versions. Once the host relaunches on a compatible + * protocol the environment is switched back on and connects normally. + */ +export const updateOutdatedHost = Effect.fn("clientRuntime.connection.updateOutdatedHost")( + function* ( + environmentId: EnvironmentId, + input: ServerSelfUpdateInput, + onStage: (stage: OutdatedHostUpdateStage) => Effect.Effect, + ) { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const resolver = yield* ConnectionResolver.ConnectionResolver; + const webSocketConstructor = yield* Socket.WebSocketConstructor; + const httpClient = yield* HttpClient.HttpClient; + const entry = (yield* SubscriptionRef.get(registry.entries)).get(environmentId); + if (entry === undefined) { + return yield* new EnvironmentRegistry.EnvironmentNotRegisteredError({ environmentId }); + } + const { prepared, descriptor } = yield* resolver.prepareForUpdate(entry); + const capabilities = descriptor.capabilities; + if ( + capabilities.serverSelfUpdate === undefined || + (capabilities.serverSelfUpdate === "desktop-managed" && + capabilities.desktopAppUpdate !== true) + ) { + return yield* new OutdatedHostUpdateError({ + environmentId, + message: `Update T3 Code on ${descriptor.label} manually; it cannot update itself.`, + }); + } + + const result = yield* Effect.scoped( + Effect.gen(function* () { + const protocolContext = yield* Layer.build( + Layer.effect( + RpcClient.Protocol, + RpcClient.makeProtocolSocket({ + retryTransientErrors: false, + retryPolicy: Schedule.recurs(0), + }), + ).pipe( + Layer.provide( + Layer.mergeAll( + Socket.layerWebSocket(prepared.socketUrl, { + openTimeout: SOCKET_OPEN_TIMEOUT, + }).pipe( + Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)), + ), + RpcSerialization.layerJson, + ), + ), + ), + ); + const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); + + const updateResult: ServerSelfUpdateResult = + capabilities.serverSelfUpdateProgress === true + ? yield* Effect.gen(function* () { + const terminal = yield* Ref.make(Option.none()); + const streamExit = yield* client[WS_METHODS.serverUpdateServerWithProgress]( + input, + ).pipe( + Stream.runForEach((event) => + event.type === "complete" + ? Ref.set(terminal, Option.some(event.result)) + : onStage(event.stage), + ), + Effect.exit, + ); + return yield* resolveServerUpdateProgressResult( + input.targetVersion, + yield* Ref.get(terminal), + streamExit, + ); + }) + : yield* client[WS_METHODS.serverUpdateServer](input).pipe( + // Older servers can drop the socket before acknowledging the restart. + Effect.catchCauseIf( + (cause) => + (capabilities.serverSelfUpdate === "boot-service" || + capabilities.serverSelfUpdate === "respawn") && + isLegacyUpdateHandoffLoss(cause), + () => + Effect.succeed({ + targetVersion: input.targetVersion, + method: capabilities.serverSelfUpdate as "boot-service" | "respawn", + } satisfies ServerSelfUpdateResult), + ), + ); + + if ( + updateResult.method === "desktop-app" && + updateResult.desktopUpdateToken !== undefined + ) { + // The commit relaunches the desktop app, so a dropped socket is success. + yield* client[WS_METHODS.serverCommitDesktopUpdate]({ + requestId: updateResult.desktopUpdateToken, + }).pipe( + Effect.catchCauseIf( + (cause) => isLegacyUpdateHandoffLoss(cause), + () => Effect.void, + ), + ); + } + return updateResult; + }), + ); + + yield* onStage("resuming"); + const resumed = yield* fetchRemoteEnvironmentDescriptor({ + httpBaseUrl: prepared.httpBaseUrl, + }).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.option, + Effect.repeat({ + schedule: Schedule.spaced(Duration.seconds(1)), + until: (current) => Option.exists(current, isCompatibleDescriptor), + }), + Effect.timeoutOption(OUTDATED_HOST_RESTART_TIMEOUT), + Effect.map(Option.flatten), + ); + if (Option.isNone(resumed)) { + return yield* new OutdatedHostUpdateError({ + environmentId, + message: `${descriptor.label} did not come back on a compatible T3 Code version.`, + }); + } + + // Discovery still holds the old relay descriptor and would re-block the + // environment from it, so replace that before clearing the block. + if (entry.target._tag === "RelayConnectionTarget") { + const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery; + yield* discovery.refresh; + } + yield* registry.setCompatibility(environmentId, null); + yield* registry.setEnabled(environmentId, true); + return { ...result, targetVersion: resumed.value.serverVersion }; + }, +); + +function isCompatibleDescriptor(descriptor: ExecutionEnvironmentDescriptor): boolean { + return (descriptor.orchestrationProtocolVersion ?? 1) === ORCHESTRATION_PROTOCOL_VERSION; +} diff --git a/packages/client-runtime/src/connection/registry.ts b/packages/client-runtime/src/connection/registry.ts index af1cc46faa9b4d28c696c86752687ac71a8f085e..ab8eeb2a3b22b92b826485a088cd850e4199b7ae 100644 --- a/packages/client-runtime/src/connection/registry.ts +++ b/packages/client-runtime/src/connection/registry.ts @@ -42,6 +42,17 @@ import { const isSshConnectionProfile = Schema.is(SshConnectionProfile); +function unsupportedState( + entry: ConnectionCatalogEntry, +): Pick { + return { + ...(entry.unsupportedReason === undefined + ? {} + : { unsupportedReason: entry.unsupportedReason }), + ...(entry.serverUpdateRequired === true ? { serverUpdateRequired: true } : {}), + }; +} + export class EnvironmentNotRegisteredError extends Schema.TaggedError()( "EnvironmentNotRegisteredError", { @@ -455,7 +466,7 @@ export const make = Effect.gen(function* () { enabled: previous.enabled, ...(previous.unsupportedReason !== undefined && gitHubRoutingConnectionKey(previous) === gitHubRoutingConnectionKey(registered) - ? { unsupportedReason: previous.unsupportedReason } + ? unsupportedState(previous) : {}), }; if ( @@ -494,7 +505,7 @@ export const make = Effect.gen(function* () { const entry: ConnectionCatalogEntry = previous?.unsupportedReason !== undefined && gitHubRoutingConnectionKey(previous) === gitHubRoutingConnectionKey(registered) - ? { ...registered, enabled: false, unsupportedReason: previous.unsupportedReason } + ? { ...registered, enabled: false, ...unsupportedState(previous) } : registered; const persistedTarget = (yield* Ref.get(persistedTargetsByEnvironment)).get( target.environmentId, @@ -844,11 +855,26 @@ export const make = Effect.gen(function* () { environmentId, Effect.gen(function* () { const entry = (yield* SubscriptionRef.get(entries)).get(environmentId); - if (entry === undefined || entry.unsupportedReason === (error?.message ?? undefined)) + if ( + entry === undefined || + (entry.unsupportedReason === (error?.message ?? undefined) && + entry.serverUpdateRequired === (error?.serverUpdateRequired ?? undefined)) + ) return; - const { unsupportedReason: _previousReason, ...rest } = entry; + const { + unsupportedReason: _previousReason, + serverUpdateRequired: _previousUpdateRequired, + ...rest + } = entry; const next: ConnectionCatalogEntry = - error === null ? rest : { ...rest, enabled: false, unsupportedReason: error.message }; + error === null + ? rest + : { + ...rest, + enabled: false, + unsupportedReason: error.message, + ...(error.serverUpdateRequired === true ? { serverUpdateRequired: true } : {}), + }; if ( error !== null && entry.enabled && diff --git a/packages/client-runtime/src/connection/resolver.ts b/packages/client-runtime/src/connection/resolver.ts index f54ebd3949fbe35a71245b7ba27be080c01f6003..77b778a50aa8fa787d12f9a26ac2fc581b5ed38a 100644 --- a/packages/client-runtime/src/connection/resolver.ts +++ b/packages/client-runtime/src/connection/resolver.ts @@ -1,4 +1,7 @@ -import type { AuthClientPresentationMetadata } from "@t3tools/contracts"; +import type { + AuthClientPresentationMetadata, + ExecutionEnvironmentDescriptor, +} from "@t3tools/contracts"; import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; @@ -49,6 +52,17 @@ export class ConnectionResolver extends Context.Service< readonly prepare: ( entry: ConnectionCatalogEntry, ) => Effect.Effect; + /** + * Authorizes a socket without the orchestration protocol gate, for hosts + * too old to connect normally. Only update RPCs may run over it. + */ + readonly prepareForUpdate: (entry: ConnectionCatalogEntry) => Effect.Effect< + { + readonly prepared: PreparedConnection; + readonly descriptor: ExecutionEnvironmentDescriptor; + }, + ConnectionAttemptError + >; } >()("@t3tools/client-runtime/connection/resolver/ConnectionResolver") {} @@ -245,7 +259,7 @@ export const make = Effect.gen(function* () { const ssh = yield* makeSshBroker(); const httpClient = yield* HttpClient.HttpClient; - const prepare = Effect.fn("clientRuntime.connection.broker.prepare")(function* ( + const authorize = Effect.fn("clientRuntime.connection.broker.authorize")(function* ( entry: ConnectionCatalogEntry, ) { const target: ConnectionTarget = entry.target; @@ -277,6 +291,13 @@ export const make = Effect.gen(function* () { actual: descriptor.environmentId, }); } + return { prepared, descriptor }; + }); + + const prepare = Effect.fn("clientRuntime.connection.broker.prepare")(function* ( + entry: ConnectionCatalogEntry, + ) { + const { prepared, descriptor } = yield* authorize(entry); const compatibilityError = orchestrationProtocolCompatibilityError(descriptor); if (compatibilityError !== null) { return yield* compatibilityError; @@ -287,7 +308,7 @@ export const make = Effect.gen(function* () { }; }); - return ConnectionResolver.of({ prepare }); + return ConnectionResolver.of({ prepare, prepareForUpdate: authorize }); }); export const layer = Layer.effect(ConnectionResolver, make); diff --git a/packages/client-runtime/src/state/outdatedServerUpdate.ts b/packages/client-runtime/src/state/outdatedServerUpdate.ts new file mode 100644 index 0000000000000000000000000000000000000000..36d7ba1f671383c29f980cf9b25a9f2822b22b07 --- /dev/null +++ b/packages/client-runtime/src/state/outdatedServerUpdate.ts @@ -0,0 +1,78 @@ +import type { EnvironmentId, ServerSelfUpdateInput } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import type * as HttpClient from "effect/unstable/http/HttpClient"; +import type { Atom } from "effect/unstable/reactivity"; +import type * as Socket from "effect/unstable/socket/Socket"; + +import { updateOutdatedHost } from "../connection/outdatedHostUpdate.ts"; +import type * as ConnectionResolver from "../connection/resolver.ts"; +import type * as EnvironmentRegistry from "../connection/registry.ts"; +import type * as RelayEnvironmentDiscovery from "../relay/discovery.ts"; +import { createAtomCommandScheduler, createRuntimeCommand } from "./runtime.ts"; +import { + serverUpdateFailureMessage, + serverUpdateStateAtom, + type ServerUpdateStage, +} from "./server.ts"; + +export interface OutdatedServerUpdateTarget { + readonly environmentId: EnvironmentId; + readonly input: ServerSelfUpdateInput; + /** From the host descriptor when known; an outdated host never delivers a server config. */ + readonly fromVersion?: string; +} + +/** + * Updates a host too old for this client to connect to. Progress lands in the + * same per-environment update state as a normal server update. + */ +export function createOutdatedServerUpdateCommand( + runtime: Atom.AtomRuntime< + | EnvironmentRegistry.EnvironmentRegistry + | ConnectionResolver.ConnectionResolver + | RelayEnvironmentDiscovery.RelayEnvironmentDiscovery + | Socket.WebSocketConstructor + | HttpClient.HttpClient, + E + >, +) { + return createRuntimeCommand(runtime, { + label: "environment-data:server:update-outdated-server", + scheduler: createAtomCommandScheduler(), + concurrency: { + mode: "singleFlight", + key: ({ environmentId }: OutdatedServerUpdateTarget) => environmentId, + }, + execute: (target: OutdatedServerUpdateTarget, atomRegistry) => { + const stateAtom = serverUpdateStateAtom(target.environmentId); + const targetVersion = target.input.targetVersion; + const fromVersion = target.fromVersion ?? targetVersion; + let currentStage: ServerUpdateStage = "downloading"; + const setStage = (stage: ServerUpdateStage) => + Effect.sync(() => { + currentStage = stage; + atomRegistry.set(stateAtom, { status: "running", stage, fromVersion, targetVersion }); + }); + return setStage(currentStage).pipe( + Effect.andThen(updateOutdatedHost(target.environmentId, target.input, setStage)), + Effect.onExit((exit) => + Effect.sync(() => { + if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) { + atomRegistry.set(stateAtom, { status: "idle" }); + return; + } + atomRegistry.set(stateAtom, { + status: "failed", + stage: currentStage, + fromVersion, + targetVersion, + message: serverUpdateFailureMessage(Cause.squash(exit.cause)), + }); + }), + ), + ); + }, + }); +} diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index a5077aa74f39e4eb3c26529388c1724b0aec1ff1..f17fa59adcc9f453451de8e9b4dc92867af8b750 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -84,7 +84,8 @@ const IDLE_SERVER_UPDATE_STATE: ServerUpdateState = { status: "idle" }; const EMPTY_SERVER_UPDATE_STATE_ATOM = Atom.make(IDLE_SERVER_UPDATE_STATE).pipe( Atom.withLabel("environment-data:server:update-state:empty"), ); -const serverUpdateStateAtom = Atom.family((environmentId: EnvironmentId) => +/** Shared with the outdated-host update, which reports through the same state. */ +export const serverUpdateStateAtom = Atom.family((environmentId: EnvironmentId) => Atom.make(IDLE_SERVER_UPDATE_STATE).pipe( Atom.withLabel(`environment-data:server:update-state:${environmentId}`), ), @@ -307,7 +308,7 @@ export function serverUpdateStateForServerVersion( : IDLE_SERVER_UPDATE_STATE; } -function serverUpdateFailureMessage(error: unknown): string { +export function serverUpdateFailureMessage(error: unknown): string { return error instanceof Error ? error.message : "Server update failed."; }