diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 5655f260bc9..b4bea21d2e3 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -32,6 +32,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverRefreshProviders]: AuthOrchestrationOperateScope, [WS_METHODS.serverUpdateProvider]: AuthOrchestrationOperateScope, [WS_METHODS.serverUpdateServer]: AuthOrchestrationOperateScope, + [WS_METHODS.serverUpdateServerWithProgress]: AuthOrchestrationOperateScope, [WS_METHODS.serverUpsertKeybinding]: AuthOrchestrationOperateScope, [WS_METHODS.serverRemoveKeybinding]: AuthOrchestrationOperateScope, [WS_METHODS.serverGetSettings]: AuthOrchestrationReadScope, diff --git a/apps/server/src/cloud/selfUpdate.test.ts b/apps/server/src/cloud/selfUpdate.test.ts index 9d6e3801704..655228c047f 100644 --- a/apps/server/src/cloud/selfUpdate.test.ts +++ b/apps/server/src/cloud/selfUpdate.test.ts @@ -26,6 +26,33 @@ import * as SelfUpdate from "./selfUpdate.ts"; const NODE_PATH = "/usr/local/bin/node"; +const eventuallyFileString = Effect.fn("test.eventuallyFileString")(function* ( + filePath: string, + expected: string, +) { + const fs = yield* FileSystem.FileSystem; + for (let iteration = 0; iteration < 1_000; iteration += 1) { + const contents = yield* fs.readFileString(filePath); + if (contents === expected) { + return; + } + // The rollback performs real filesystem I/O on a detached fiber, which + // advancing TestClock does not await. + yield* Effect.yieldNow; + } + return yield* Effect.die(new Error(`Expected file contents were not observed at ${filePath}.`)); +}); + +const eventuallyTrue = Effect.fn("test.eventuallyTrue")(function* (predicate: () => boolean) { + for (let iteration = 0; iteration < 1_000; iteration += 1) { + if (predicate()) { + return; + } + yield* Effect.yieldNow; + } + return yield* Effect.die(new Error("Expected condition was not observed.")); +}); + interface RecordedCommand { readonly command: string; readonly args: ReadonlyArray; @@ -457,8 +484,12 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { it.effect("installs, preflights, and respawns a foreground server", () => Effect.gen(function* () { const context = yield* makeContext(); - const result = yield* context.service.update({ targetVersion: "0.0.29" }); + const progress: Array = []; + const result = yield* context.service.update({ targetVersion: "0.0.29" }, (stage) => + Effect.sync(() => progress.push(stage)), + ); assert.deepEqual(result, { targetVersion: "0.0.29", method: "respawn" }); + assert.deepEqual(progress, ["downloading", "installing"]); assert.lengthOf(context.spawns, 1); const concurrentError = yield* context.service @@ -506,10 +537,12 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { assert.include(unit, `ExecStart=${NODE_PATH} ${pinnedEntry} serve`); assert.deepEqual( context.commands.map((entry) => entry.command), - ["npm", NODE_PATH, "systemctl", "systemctl"], + ["npm", NODE_PATH, "systemctl"], ); assert.deepEqual(context.commands[2]?.args, ["--user", "daemon-reload"]); + // Restart waits until after the update acknowledgement can flush. + yield* TestClock.adjust(Duration.seconds(10)); assert.deepEqual(context.commands[3], { command: "systemctl", args: ["--user", "restart", "t3code.service"], @@ -542,9 +575,11 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { ); const previousUnit = yield* context.fs.readFileString(unitPath); - const first = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip); - assert.include(first.reason, "Restarting the systemd boot service failed"); - assert.equal(yield* context.fs.readFileString(unitPath), previousUnit); + const first = yield* context.service.update({ targetVersion: "0.0.29" }); + assert.deepEqual(first, { targetVersion: "0.0.29", method: "boot-service" }); + yield* TestClock.adjust(Duration.seconds(10)); + yield* eventuallyFileString(unitPath, previousUnit); + yield* eventuallyTrue(() => context.commands.at(-1)?.args[1] === "daemon-reload"); assert.deepEqual( context.commands.slice(-2).map((entry) => entry.args), [ diff --git a/apps/server/src/cloud/selfUpdate.ts b/apps/server/src/cloud/selfUpdate.ts index 62bd07fbbc8..f786cbea8cf 100644 --- a/apps/server/src/cloud/selfUpdate.ts +++ b/apps/server/src/cloud/selfUpdate.ts @@ -6,6 +6,7 @@ import { ServerSelfUpdateError, type ServerSelfUpdateCapability, type ServerSelfUpdateInput, + type ServerSelfUpdateProgressStage, type ServerSelfUpdateResult, } from "@t3tools/contracts"; import { @@ -147,6 +148,7 @@ export class ServerSelfUpdate extends Context.Service< { readonly update: ( input: ServerSelfUpdateInput, + reportProgress?: (stage: ServerSelfUpdateProgressStage) => Effect.Effect, ) => Effect.Effect; } >()("t3/cloud/selfUpdate/ServerSelfUpdate") {} @@ -227,7 +229,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option const update: ServerSelfUpdate["Service"]["update"] = Effect.fn( "cloud.server_self_update.update", - )(function* (input) { + )(function* (input, reportProgress = () => Effect.void) { if (capability === "desktop-managed") { return yield* failWith( "This server is managed by the T3 Code desktop app on its machine; update the desktop app to update it.", @@ -250,6 +252,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option } return yield* Effect.gen(function* () { + yield* reportProgress("downloading"); const runtimePaths = yield* ensurePinnedRuntimeInstalled({ baseDir: serverConfig.baseDir, version: targetVersion, @@ -260,6 +263,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option Effect.mapError((error) => failWith("Could not install the requested t3 version.", error)), ); + yield* reportProgress("installing"); // A broken artifact (failed native build, incompatible node) must be // caught while the current server is still alive to report it. const preflight = yield* runner @@ -349,38 +353,45 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option yield* Effect.logInfo("Server self-update installed; restarting boot service.", { targetVersion, }); - // A successful systemd restart stops this process, so the RPC is - // interrupted and the reconnecting client observes the new version. - // A rejected restart returns while the old process is still alive; - // restore the previous unit and report that failure through the RPC. - yield* Effect.gen(function* () { - const restart = yield* runner - .run({ - command: "systemctl", - args: ["--user", "restart", BOOT_SERVICE_UNIT_FILE], - }) - .pipe( - Effect.mapError((cause) => - failWith("Could not restart the systemd boot service.", cause), + // Restart after the acknowledgement has had time to cross any relay + // hop. If systemd rejects the handoff, restore the previous unit while + // this process is still alive and log the failure for diagnostics. + yield* scheduleRestart( + Effect.gen(function* () { + const restart = yield* runner + .run({ + command: "systemctl", + args: ["--user", "restart", BOOT_SERVICE_UNIT_FILE], + }) + .pipe( + Effect.mapError((cause) => + failWith("Could not restart the systemd boot service.", cause), + ), + ); + if (restart.code !== 0) { + return yield* failWith( + `Restarting the systemd boot service failed (exit code ${String(restart.code)}).`, + ); + } + }).pipe( + Effect.catch((restartError) => + writeUnitAtomically(unitPath, previousUnit).pipe( + Effect.andThen(reloadSystemd()), + Effect.mapError((rollbackError) => + failWith("Could not restore the previous systemd unit.", { + restartError, + rollbackError, + }), + ), + Effect.andThen(Effect.fail(restartError)), ), - ); - if (restart.code !== 0) { - return yield* failWith( - `Restarting the systemd boot service failed (exit code ${String(restart.code)}).`, - ); - } - }).pipe( - Effect.catch((restartError) => - writeUnitAtomically(unitPath, previousUnit).pipe( - Effect.andThen(reloadSystemd()), - Effect.mapError((rollbackError) => - failWith("Could not restore the previous systemd unit.", { - restartError, - rollbackError, - }), + ), + Effect.catch((error) => + Effect.logError("Server self-update could not restart the boot service.").pipe( + Effect.annotateLogs({ targetVersion, error: error.reason }), ), - Effect.andThen(Effect.fail(restartError)), ), + Effect.ensuring(Ref.set(inFlight, false)), ), ); } else { diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 0eaf5a7c16a..dbf3651c229 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -143,6 +143,9 @@ export const make = Effect.gen(function* () { threadSettlement: true, threadSnooze: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), + ...(serverSelfUpdate === "boot-service" || serverSelfUpdate === "respawn" + ? { serverSelfUpdateProgress: true } + : {}), }, }; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 6c021c9af80..e33c6992bbb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -40,6 +40,8 @@ import { ProjectWriteFileError, RelayClientInstallFailedError, type RelayClientInstallProgressEvent, + type ServerSelfUpdateError, + type ServerSelfUpdateProgressEvent, type FilesystemBrowseFailure, FilesystemBrowseError, AssetWorkspaceContextNotFoundError, @@ -1364,6 +1366,33 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.serverUpdateServer, serverSelfUpdate.update(input), { "rpc.aggregate": "server", }), + [WS_METHODS.serverUpdateServerWithProgress]: (input) => + observeRpcStream( + WS_METHODS.serverUpdateServerWithProgress, + Stream.callback((queue) => + serverSelfUpdate + .update(input, (stage) => + Queue.offer(queue, { + type: "progress", + stage, + }).pipe(Effect.asVoid), + ) + .pipe( + Effect.flatMap((result) => + Queue.offer(queue, { + type: "complete", + result, + }), + ), + Effect.catchTags({ + ServerSelfUpdateError: (error) => Queue.fail(queue, error), + }), + Effect.andThen(Queue.end(queue)), + Effect.forkScoped, + ), + ), + { "rpc.aggregate": "server" }, + ), [WS_METHODS.serverUpsertKeybinding]: (rule) => observeRpcEffect( WS_METHODS.serverUpsertKeybinding, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index d532c8b1233..ec85e4c4c18 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -296,7 +296,7 @@ import { AlertDialogTitle, } from "./ui/alert-dialog"; import { Tooltip, TooltipPopup, TooltipTrigger } from "./ui/tooltip"; -import { ServerUpdateAction } from "./ServerUpdateAction"; +import { ServerUpdateAction, ServerUpdateProgress } from "./ServerUpdateAction"; import { buildVersionMismatchDismissalKey, dismissVersionMismatch, @@ -1876,12 +1876,16 @@ function ChatViewContent(props: ChatViewProps) { hasMultipleRegisteredEnvironments && activeThread ? `${environmentById.get(activeThread.environmentId)?.label ?? serverConfig?.environment.label ?? activeThread.environmentId} server` : "server"; - const versionMismatchEnvironmentId = - versionMismatch && activeThread ? activeThread.environmentId : null; + const serverUpdateEnvironmentId = activeThread?.environmentId ?? null; const versionMismatchSelfUpdate = resolveServerSelfUpdateCapability(serverConfig); + const serverUpdateState = useAtomValue( + serverEnvironment.updateStateAtom(serverUpdateEnvironmentId), + ); const systemComposerBannerItems = useMemo(() => { const items: ComposerBannerStackItem[] = []; - if (activeEnvironmentUnavailableState) { + const resumingServerUpdate = + serverUpdateState.status === "running" && serverUpdateState.stage === "resuming"; + if (activeEnvironmentUnavailableState && !resumingServerUpdate) { const connection = activeEnvironmentUnavailableState.connection; const isReconnecting = connection.phase === "connecting" || connection.phase === "reconnecting"; @@ -1918,39 +1922,57 @@ function ChatViewContent(props: ChatViewProps) { }); } if ( - showVersionMismatchBanner && - versionMismatch && - versionMismatchDismissKey && - versionMismatchEnvironmentId + serverUpdateEnvironmentId && + (serverUpdateState.status !== "idle" || + (showVersionMismatchBanner && versionMismatch && versionMismatchDismissKey)) ) { + const updateInProgress = serverUpdateState.status === "running"; + const updateFailed = serverUpdateState.status === "failed"; items.push({ - id: `version-mismatch:${versionMismatchDismissKey}`, - variant: "warning", + id: `server-version:${serverUpdateEnvironmentId}`, + variant: updateFailed ? "error" : "warning", icon: , - title: "Client and server versions differ", - description: ( - <> - Client {versionMismatch.clientVersion} is connected to {versionMismatchServerLabel}{" "} - {versionMismatch.serverVersion}.{" "} - {serverUpdateGuidance(versionMismatchSelfUpdate, versionMismatchServerLabel)} - - ), + title: + updateInProgress || updateFailed + ? `${updateFailed ? "Could not update" : "Updating"} ${versionMismatchServerLabel}` + : "Client and server versions differ", + description: + updateInProgress || updateFailed ? ( + + ) : versionMismatch ? ( + <> + Client {versionMismatch.clientVersion} is connected to {versionMismatchServerLabel}{" "} + {versionMismatch.serverVersion}.{" "} + {serverUpdateGuidance(versionMismatchSelfUpdate, versionMismatchServerLabel)} + + ) : null, // The desktop-managed guidance is already the description; the action // slot would only repeat it. actions: + updateInProgress || + !versionMismatch || versionMismatchSelfUpdate === "desktop-managed" ? undefined : ( ), - dismissLabel: "Dismiss version mismatch warning", - onDismiss: () => { - dismissVersionMismatch(versionMismatchDismissKey); - setDismissedVersionMismatchKey(versionMismatchDismissKey); - }, + ...(updateInProgress || updateFailed || !versionMismatchDismissKey + ? {} + : { + dismissLabel: "Dismiss version mismatch warning", + onDismiss: () => { + dismissVersionMismatch(versionMismatchDismissKey); + setDismissedVersionMismatchKey(versionMismatchDismissKey); + }, + }), }); } return items; @@ -1960,9 +1982,10 @@ function ChatViewContent(props: ChatViewProps) { navigate, setDismissedVersionMismatchKey, showVersionMismatchBanner, + serverUpdateState, versionMismatch, versionMismatchDismissKey, - versionMismatchEnvironmentId, + serverUpdateEnvironmentId, versionMismatchSelfUpdate, versionMismatchServerLabel, ]); diff --git a/apps/web/src/components/ServerUpdateAction.test.tsx b/apps/web/src/components/ServerUpdateAction.test.tsx index c0ba1c693d4..17e44b8bae6 100644 --- a/apps/web/src/components/ServerUpdateAction.test.tsx +++ b/apps/web/src/components/ServerUpdateAction.test.tsx @@ -1,71 +1,15 @@ -import type { Dispatch, ReactElement, SetStateAction } from "react"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import type { ReactElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; +import type { EnvironmentId } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import { AsyncResult } from "effect/unstable/reactivity"; -import type { EnvironmentId } from "@t3tools/contracts"; +import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; const testState = vi.hoisted(() => ({ updateServer: vi.fn(), toast: vi.fn(), })); -const hooks = vi.hoisted(() => { - let cursor = 0; - let slots: unknown[] = []; - const nextIndex = () => cursor++; - - return { - beginRender() { - cursor = 0; - }, - reset() { - cursor = 0; - slots = []; - }, - useEffect() { - nextIndex(); - }, - useMemoCache(size: number): unknown[] { - const index = nextIndex(); - if (!slots[index]) { - slots[index] = Array.from({ length: size }, () => Symbol.for("react.memo_cache_sentinel")); - } - return slots[index] as unknown[]; - }, - useRef(initialValue: T): { current: T } { - const index = nextIndex(); - if (!slots[index]) { - slots[index] = { current: initialValue }; - } - return slots[index] as { current: T }; - }, - useState(initialValue: T | (() => T)): [T, Dispatch>] { - const index = nextIndex(); - if (index >= slots.length) { - slots[index] = - typeof initialValue === "function" ? (initialValue as () => T)() : initialValue; - } - const setValue: Dispatch> = (nextValue) => { - const previous = slots[index] as T; - slots[index] = - typeof nextValue === "function" ? (nextValue as (value: T) => T)(previous) : nextValue; - }; - return [slots[index] as T, setValue]; - }, - }; -}); - -vi.mock("react", async (importOriginal) => { - const actual = await importOriginal(); - return { - ...actual, - useEffect: hooks.useEffect, - useRef: hooks.useRef, - useState: hooks.useState, - }; -}); - -vi.mock("react/compiler-runtime", () => ({ c: hooks.useMemoCache })); vi.mock("~/hooks/useCopyToClipboard", () => ({ useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }), })); @@ -79,120 +23,123 @@ vi.mock("./ui/toast", () => ({ toastManager: { add: testState.toast }, })); -import { ServerUpdateAction } from "./ServerUpdateAction"; +import { ServerUpdateAction, ServerUpdateProgress } from "./ServerUpdateAction"; type ActionElement = ReactElement<{ - readonly disabled?: boolean; readonly onClick?: () => void; }>; function renderAction(): ActionElement { - hooks.beginRender(); return ServerUpdateAction({ environmentId: "env-test" as EnvironmentId, serverLabel: "Test server", selfUpdate: "boot-service", - targetVersion: "0.0.29", + targetVersion: "0.0.31", }) as ActionElement; } -function deferred() { - let resolve!: (value: T) => void; - const promise = new Promise((complete) => { - resolve = complete; - }); - return { promise, resolve }; -} - async function flushPromises(): Promise { await Promise.resolve(); await Promise.resolve(); - await Promise.resolve(); - await Promise.resolve(); - await Promise.resolve(); } describe("ServerUpdateAction", () => { beforeEach(() => { - vi.useFakeTimers(); - hooks.reset(); testState.updateServer.mockReset(); testState.toast.mockReset(); }); - afterEach(() => { - vi.useRealTimers(); - }); - - it("starts a fresh reconnect timeout after a long install succeeds", async () => { - const update = - deferred< - ReturnType> - >(); - testState.updateServer.mockReturnValue(update.promise); + it("reports success only after the shared update flow reconnects", async () => { + testState.updateServer.mockResolvedValue( + AsyncResult.success({ targetVersion: "0.0.31", method: "boot-service" as const }), + ); renderAction().props.onClick?.(); - expect(renderAction().props.disabled).toBe(true); - - await vi.advanceTimersByTimeAsync(11 * 60_000); - update.resolve( - AsyncResult.success({ - targetVersion: "0.0.29", - method: "boot-service", - }), - ); await flushPromises(); - // The click-based deadline would have fired by now. Success gets a fresh - // twelve-minute reconnect window, so the action remains disabled. - await vi.advanceTimersByTimeAsync(2 * 60_000); - expect(renderAction().props.disabled).toBe(true); - expect(testState.toast).not.toHaveBeenCalledWith( - expect.objectContaining({ title: "Server update timed out" }), - ); - - await vi.advanceTimersByTimeAsync(10 * 60_000); - expect(renderAction().props.disabled).not.toBe(true); - expect(testState.toast).toHaveBeenCalledWith( - expect.objectContaining({ title: "Server update timed out" }), - ); + expect(testState.updateServer).toHaveBeenCalledWith({ + environmentId: "env-test", + input: { targetVersion: "0.0.31" }, + }); + expect(testState.toast).toHaveBeenCalledWith({ + type: "success", + title: "Test server updated", + description: "Reconnected on t3@0.0.31.", + }); }); - it("does not let an expired request clear a newer retry", async () => { - const first = deferred>(); - const retry = - deferred< - ReturnType> - >(); - testState.updateServer.mockReturnValueOnce(first.promise).mockReturnValueOnce(retry.promise); - - renderAction().props.onClick?.(); - await vi.advanceTimersByTimeAsync(12 * 60_000); - expect(renderAction().props.disabled).not.toBe(true); - - renderAction().props.onClick?.(); - expect(renderAction().props.disabled).toBe(true); - - first.resolve(AsyncResult.failure(Cause.fail(new Error("first request failed late")))); - await flushPromises(); + it("reports one result when the update action is double-clicked", async () => { + let finishUpdate: (() => void) | undefined; + testState.updateServer.mockImplementation( + () => + new Promise((resolve) => { + finishUpdate = () => + resolve( + AsyncResult.success({ targetVersion: "0.0.31", method: "boot-service" as const }), + ); + }), + ); - expect(renderAction().props.disabled).toBe(true); - expect(testState.updateServer).toHaveBeenCalledTimes(2); + const action = renderAction(); + action.props.onClick?.(); + action.props.onClick?.(); - retry.resolve(AsyncResult.success({ targetVersion: "0.0.29", method: "boot-service" })); + expect(testState.updateServer).toHaveBeenCalledTimes(1); + finishUpdate?.(); await flushPromises(); - expect(renderAction().props.disabled).toBe(true); + expect(testState.toast).toHaveBeenCalledTimes(1); }); - it("quietly releases the action when a restart RPC is interrupted", async () => { + it("quietly releases the action when the operation is interrupted", async () => { testState.updateServer.mockResolvedValue(AsyncResult.failure(Cause.interrupt())); renderAction().props.onClick?.(); await flushPromises(); - expect(renderAction().props.disabled).not.toBe(true); - expect(testState.toast).not.toHaveBeenCalledWith( - expect.objectContaining({ title: "Server update failed" }), + expect(testState.toast).not.toHaveBeenCalled(); + }); +}); + +describe("ServerUpdateProgress", () => { + it("renders the chosen horizontal step rail without an animated spinner", () => { + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain("0.0.30"); + expect(markup).toContain("0.0.31"); + expect(markup).toContain("Download"); + expect(markup).toContain("Install"); + expect(markup).toContain("Resume"); + expect(markup).toContain("Waiting for bb-1 to accept commands."); + expect(markup).not.toContain("animate-spin"); + }); + + it("keeps the failed stage visible with its retryable error", () => { + const markup = renderToStaticMarkup( + , ); + + expect(markup).toContain('role="alert"'); + expect(markup).toContain("The package could not be verified."); }); }); diff --git a/apps/web/src/components/ServerUpdateAction.tsx b/apps/web/src/components/ServerUpdateAction.tsx index 5e82c4e4d93..929ea2e56c7 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -1,55 +1,147 @@ -import { useEffect, useRef, useState } from "react"; import type { EnvironmentId, ServerSelfUpdateCapability } from "@t3tools/contracts"; +import type { ServerUpdateState } from "@t3tools/client-runtime/state/server"; import { isAtomCommandInterrupted, squashAtomCommandFailure, } from "@t3tools/client-runtime/state/runtime"; +import { CheckIcon } from "lucide-react"; import { useCopyToClipboard } from "~/hooks/useCopyToClipboard"; +import { cn } from "~/lib/utils"; import { serverEnvironment } from "~/state/server"; import { useAtomCommand } from "~/state/use-atom-command"; import { manualServerUpdateCommand } from "~/versionSkew"; import { Button } from "./ui/button"; -import { Spinner } from "./ui/spinner"; import { toastManager } from "./ui/toast"; -/** - * The npm install on the server side is capped at 10 minutes; expire the - * spinner a bit beyond that so a dead transport never strands a disabled - * button, while a legitimately slow install is never cut off. - */ -const UPDATE_PENDING_EXPIRY_MS = 12 * 60_000; +const UPDATE_STEPS = [ + { stage: "downloading", label: "Download" }, + { stage: "installing", label: "Install" }, + { stage: "resuming", label: "Resume" }, +] as const; +const pendingUpdateEnvironmentIds = new Set(); function updateFailureMessage(error: unknown): string { return error instanceof Error ? error.message : "Server update failed."; } +function updateStatusCopy( + state: Exclude, + serverLabel: string, +): string { + if (state.status === "failed") { + return state.message; + } + switch (state.stage) { + case "downloading": + return "Downloading the matching server version."; + case "installing": + return `Installing and verifying t3@${state.targetVersion}.`; + case "resuming": + return `Waiting for ${serverLabel} to accept commands.`; + } +} + +export function ServerUpdateProgress({ + fromVersion, + serverLabel, + state, +}: { + readonly fromVersion: string; + readonly serverLabel: string; + readonly state: Exclude; +}) { + const currentIndex = UPDATE_STEPS.findIndex(({ stage }) => stage === state.stage); + + return ( +
+

+ {fromVersion} {state.targetVersion} +

+
    + {UPDATE_STEPS.map((step, index) => { + const complete = index < currentIndex; + const current = index === currentIndex; + const failed = current && state.status === "failed"; + return ( +
  1. + + {step.label} + {index < UPDATE_STEPS.length - 1 ? ( +
    2. + ); + })} +
+

+ {updateStatusCopy(state, serverLabel)} +

+
+ ); +} + /** - * The call-to-action for a version-skewed server, matched to the update path - * it advertises: a one-click install-and-restart for servers that can update - * themselves, an update-the-desktop-app hint for desktop-managed backends - * (running `npx t3` there would start a second server, not update this one), - * and copying the manual relaunch command for everything else — so the skew - * warning always offers a way out. + * Offers the update path advertised by a version-skewed server. Self-updates + * delegate their full lifecycle to client-runtime so this component can + * unmount during reconnect without losing operation state. */ export function ServerUpdateAction({ environmentId, serverLabel, selfUpdate, targetVersion, + label = "Update server", }: { readonly environmentId: EnvironmentId; readonly serverLabel: string; readonly selfUpdate: ServerSelfUpdateCapability | null; readonly targetVersion: string; + readonly label?: string; }) { const updateServer = useAtomCommand(serverEnvironment.updateServer, { reportFailure: false, }); - const [pending, setPending] = useState(false); - const inFlightRef = useRef(false); - const attemptRef = useRef(0); - const expiryRef = useRef | null>(null); const { copyToClipboard } = useCopyToClipboard<{ command: string }>({ target: "update command", onCopy: ({ command }) => { @@ -68,107 +160,35 @@ export function ServerUpdateAction({ }, }); - useEffect( - () => () => { - if (expiryRef.current !== null) { - clearTimeout(expiryRef.current); - expiryRef.current = null; - } - attemptRef.current += 1; - inFlightRef.current = false; - }, - [], - ); - - const handleUpdate = () => { - // Synchronous re-entry guard: setPending is async, so a rapid - // double-click would otherwise dispatch two updates. - if (inFlightRef.current) { + const handleUpdate = async () => { + if (pendingUpdateEnvironmentIds.has(environmentId)) { return; } - inFlightRef.current = true; - const attempt = attemptRef.current + 1; - attemptRef.current = attempt; - const ownsAttempt = () => attemptRef.current === attempt; - setPending(true); - const armExpiry = () => { - const expiry = setTimeout(() => { - if (!ownsAttempt()) return; - expiryRef.current = null; - attemptRef.current += 1; - inFlightRef.current = false; - setPending(false); - toastManager.add({ - type: "error", - title: "Server update timed out", - description: "The update may still be running on the server — check again in a minute.", - }); - }, UPDATE_PENDING_EXPIRY_MS); - expiryRef.current = expiry; - return expiry; - }; - let expiry = armExpiry(); - let restartAccepted = false; - const keepPendingForRestart = () => { - restartAccepted = true; - if (expiryRef.current === expiry) { - clearTimeout(expiry); - expiry = armExpiry(); - } - }; - void Promise.resolve() - .then(() => - updateServer({ - environmentId, - input: { targetVersion }, - }), - ) - .then((result) => { - if (!ownsAttempt()) return; - if (result._tag === "Failure") { - // An interrupt may be the expected boot-service disconnect, but it - // can also be client-side cancellation before restart was accepted. - // Release the action quietly; version sync will remove it when a - // successful replacement reconnects. - if (isAtomCommandInterrupted(result)) { - return; - } - toastManager.add({ - type: "error", - title: "Server update failed", - description: updateFailureMessage(squashAtomCommandFailure(result)), - }); + pendingUpdateEnvironmentIds.add(environmentId); + try { + const result = await updateServer({ + environmentId, + input: { targetVersion }, + }); + if (result._tag === "Failure") { + if (isAtomCommandInterrupted(result)) { return; } - keepPendingForRestart(); - // Installation can legitimately consume most of the request window. - // Give restart/reconnect a fresh full window after the server accepts - // the handoff instead of expiring based on the original click time. - toastManager.add({ - type: "success", - title: `Updating ${serverLabel}`, - description: `t3@${result.value.targetVersion} is installed — the server is restarting and will reconnect shortly.`, - }); - }) - .catch((error: unknown) => { - if (!ownsAttempt()) return; toastManager.add({ type: "error", title: "Server update failed", - description: updateFailureMessage(error), + description: updateFailureMessage(squashAtomCommandFailure(result)), }); - }) - .finally(() => { - // A successful RPC only acknowledges that restart is scheduled. Keep - // the action disabled until version sync unmounts it, or until the - // safety expiry reports that reconnection never arrived. - if (restartAccepted || !ownsAttempt() || expiryRef.current !== expiry) return; - expiryRef.current = null; - clearTimeout(expiry); - attemptRef.current += 1; - inFlightRef.current = false; - setPending(false); + return; + } + toastManager.add({ + type: "success", + title: `${serverLabel} updated`, + description: `Reconnected on t3@${result.value.targetVersion}.`, }); + } finally { + pendingUpdateEnvironmentIds.delete(environmentId); + } }; if (selfUpdate === "desktop-managed") { @@ -188,14 +208,9 @@ export function ServerUpdateAction({ ); } - return pending ? ( - - ) : ( + return ( ); } diff --git a/apps/web/src/components/settings/ConnectionsSettings.tsx b/apps/web/src/components/settings/ConnectionsSettings.tsx index 632221098d2..98ab7a128de 100644 --- a/apps/web/src/components/settings/ConnectionsSettings.tsx +++ b/apps/web/src/components/settings/ConnectionsSettings.tsx @@ -7,6 +7,7 @@ import { TerminalIcon, TriangleAlertIcon, } from "lucide-react"; +import { useAtomValue } from "@effect/atom-react"; import { type ReactNode, memo, useCallback, useMemo, useState } from "react"; import { AuthAccessReadScope, @@ -131,8 +132,9 @@ import { usePrimaryEnvironment, } from "~/state/environments"; import { useAtomCommand } from "../../state/use-atom-command"; +import { serverEnvironment } from "~/state/server"; import { ConnectionStatusDot } from "../ConnectionStatusDot"; -import { ServerUpdateAction } from "../ServerUpdateAction"; +import { ServerUpdateAction, ServerUpdateProgress } from "../ServerUpdateAction"; import { CloudEnvironmentConnectRows } from "../cloud/CloudEnvironmentConnectList"; import { ITEM_ROW_CLASSNAME, ITEM_ROW_INNER_CLASSNAME } from "./itemRows"; @@ -1389,6 +1391,9 @@ function SavedBackendListRow({ [copyTraceIdToClipboard], ); const versionMismatch = resolveServerConfigVersionMismatch(environment.serverConfig); + const serverUpdateState = useAtomValue(serverEnvironment.updateStateAtom(environmentId)); + const resumingServerUpdate = + serverUpdateState.status === "running" && serverUpdateState.stage === "resuming"; const sshTarget = environment.entry.target._tag === "SshConnectionTarget" && Option.isSome(environment.entry.profile) && @@ -1425,14 +1430,22 @@ function SavedBackendListRow({ {metadataBits.length > 0 ? (

{metadataBits.join(" · ")}

) : null} - {versionMismatch ? ( + {serverUpdateState.status !== "idle" ? ( +
+ +
+ ) : versionMismatch ? (

Version drift: client {versionMismatch.clientVersion}, server{" "} {versionMismatch.serverVersion}.

) : null} - {environment.connection.error ? ( + {environment.connection.error && !resumingServerUpdate ? (

{connectionStatusText(environment.connection)} {errorTraceId ? ( @@ -1448,12 +1461,14 @@ function SavedBackendListRow({ ) : null}

- {versionMismatch ? ( + {versionMismatch && + (serverUpdateState.status === "idle" || serverUpdateState.status === "failed") ? ( ) : null} {isWslEnvironment ? ( @@ -1834,6 +1849,9 @@ export function ConnectionsSettings() { >(null); const primaryServerConfig = primaryEnvironment?.serverConfig ?? null; const primaryVersionMismatch = resolveServerConfigVersionMismatch(primaryServerConfig); + const primaryServerUpdateState = useAtomValue( + serverEnvironment.updateStateAtom(primaryEnvironmentId), + ); const [isAdvertisedEndpointListExpanded, setIsAdvertisedEndpointListExpanded] = useState(false); const defaultAdvertisedEndpointKey = useUiStateStore( (state) => state.defaultAdvertisedEndpointKey, @@ -2980,24 +2998,43 @@ export function ConnectionsSettings() { {canManageLocalBackend ? ( <> - {primaryVersionMismatch ? ( + {primaryVersionMismatch || primaryServerUpdateState.status !== "idle" ? ( - - Client {primaryVersionMismatch.clientVersion}, server{" "} - {primaryVersionMismatch.serverVersion}. Sync them if RPC calls or reconnects - fail. - + primaryServerUpdateState.status !== "idle" ? ( + + ) : primaryVersionMismatch ? ( + + + Client {primaryVersionMismatch.clientVersion}, server{" "} + {primaryVersionMismatch.serverVersion}. Sync them if RPC calls or reconnects + fail. + + ) : null } control={ - primaryEnvironmentId !== null ? ( + primaryVersionMismatch && + primaryEnvironmentId !== null && + primaryServerUpdateState.status !== "running" ? ( ) : undefined } diff --git a/docs/architecture/server-updates.md b/docs/architecture/server-updates.md index 2e8c9d6ef61..981f4e1eb42 100644 --- a/docs/architecture/server-updates.md +++ b/docs/architecture/server-updates.md @@ -13,8 +13,9 @@ The feature has three boundaries: ## Detection and Presentation `ExecutionEnvironmentDescriptor` includes the server version and an optional -`capabilities.serverSelfUpdate` value. The client compares that version with `APP_VERSION` after -loading server config. +`capabilities.serverSelfUpdate` value. Progress-capable servers also advertise +`capabilities.serverSelfUpdateProgress`. The client compares the server version with `APP_VERSION` +after loading server config. The optional capability is intentionally backward compatible. An older server does not know about the field, so a missing value means the client must offer a manual relaunch instead of sending an @@ -28,6 +29,10 @@ The shared `ServerUpdateAction` is rendered in both user-facing version-drift su Both surfaces target the client's exact version. When the reconnected server reports that version, the mismatch and action disappear. +The operation state lives in `packages/client-runtime`, keyed by environment. Both web surfaces read +the same `downloading`, `installing`, or `resuming` state, so route changes do not own or cancel the +operation. + ## Capability Selection The server resolves its capability once at startup and publishes it in the environment descriptor. @@ -51,20 +56,25 @@ flowchart TD A[Client detects different versions] --> B{Advertised update path} B -->|desktop-managed| C[Update desktop app on server machine] B -->|missing| D[Copy exact manual relaunch command] - B -->|boot-service or respawn| E[server.updateServer] - E --> F[Install exact t3 version in pinned runtime] - F --> G[Run version preflight] - G -->|fails| H[Remove failed runtime and keep current server] - G -->|passes| I{Handoff method} - I -->|boot-service| J[Rewrite and restart T3 systemd unit] - I -->|respawn| K[Start delayed replacement and exit current process] - J --> L[Client reconnects] - K --> L + B -->|boot-service or respawn| E{Progress capability} + E -->|present| F[server.updateServerWithProgress] + E -->|missing| G[server.updateServer fallback] + F --> H[Download exact t3 version] + G --> H + H --> I[Install and run version preflight] + I -->|fails| J[Remove failed runtime and keep current server] + I -->|passes| K{Handoff method} + K -->|boot-service| L[Rewrite and restart T3 systemd unit] + K -->|respawn| M[Start delayed replacement and exit current process] + L --> N[Reconnect with fresh backoff] + M --> N + N --> O[Replacement publishes ready at target version] ``` -`server.updateServer` requires the environment's `orchestration:operate` authorization scope. Its +Both update RPCs require the environment's `orchestration:operate` authorization scope. Their payload accepts only an exact npm version, including an exact prerelease version; dist-tags such as -`latest` and `nightly` are rejected. +`latest` and `nightly` are rejected. The unary `server.updateServer` method remains available so a +new client can still repair skew with an older server. The update service permits one update at a time. It installs `t3@` under `/runtime/versions/` and writes an install-complete sentinel only after npm exits @@ -89,20 +99,20 @@ authorization; it does not uninstall the host service. ## Process Handoff For `boot-service`, the server atomically rewrites the T3-managed user unit to point at the verified -runtime, reloads systemd, and restarts the unit. Reload and restart failures restore the previous -unit before returning an error. +runtime and reloads systemd. It acknowledges the handoff, then restarts the unit after the same +short grace period used by foreground respawn. A rejected deferred restart restores the previous +unit and is logged by the still-running process. For `respawn`, the server starts a detached, delayed replacement that replays the original CLI arguments. It then acknowledges the request and schedules the current process to exit. The delays give the acknowledgement time to cross direct or relayed connections before the socket closes. -There is no separate progress stream. The update request remains pending while npm installs and the -client shows a disabled update action. A restart can interrupt the request normally; the connection -runtime keeps the environment registered and reconnects through its usual retry path. After an -acknowledged foreground handoff, the UI keeps the action pending until version sync removes it or a -safety timeout releases it. If a boot-service restart closes the connection before acknowledgement, -the UI releases the interrupted action without reporting a false update failure and lets reconnect -and the next version check determine the result. +Progress-capable servers emit `downloading` before installing the pinned runtime and `installing` +before preflight and handoff. A terminal stream event acknowledges that restart is scheduled. The +client then enters `resuming`, waits for the replacement lifecycle stream to publish `ready` with +the target version, and only then completes the operation. It watches for the intentional +disconnect's first backoff state and requests one fresh retry, which clears historical backoff debt +without adding a separate reconnect loop. ## Release Invariant diff --git a/docs/user/server-updates.md b/docs/user/server-updates.md index 6a8e33977b7..27577c6f948 100644 --- a/docs/user/server-updates.md +++ b/docs/user/server-updates.md @@ -31,6 +31,11 @@ The update does not remove saved threads, settings, or project files. The available action depends on how that server was started. T3 Code does not update connected servers silently in the background. +After selecting **Update server**, the warning becomes a three-step progress rail: +**Download**, **Install**, and **Resume**. The same progress appears in the conversation and in +Connections, so navigating between them does not lose the update. A failed step remains visible +with its error and an option to retry. + If the server uses the T3 Code background service, you can also update it directly on the host: ```sh @@ -42,11 +47,11 @@ commands. ## After the Update -Keep the web or desktop app open while the server restarts. When it reconnects with the matching -version, the warning and update action disappear. +Keep the web or desktop app open while the server restarts. The update completes only after the +replacement server reports the requested version and is ready to accept commands. The warning and +progress rail then disappear. -If the client reports a timeout, the server may still be finishing the update. Wait a minute, then -reconnect or open **Settings** → **Connections** again. If the warning remains: +If a step fails: 1. Retry the offered action once. 2. Make sure you updated the machine named in the warning, not only the device you are using. diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f3901e42251..660ae21efd4 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -481,6 +481,43 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("explicit retry starts a fresh backoff sequence", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: () => Effect.fail(transient()), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + + yield* supervisor.retryNow; + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(3); + + yield* TestClock.adjust("999 millis"); + expect(yield* Ref.get(harness.prepareCount)).toBe(3); + yield* TestClock.adjust("1 milli"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(4); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("keeps blocked failures idle until an external signal requests another attempt", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d9efcd4263a..f64d60b276d 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -229,6 +229,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }; const intent = yield* Ref.make(initialIntent); const signals = yield* Queue.unbounded(); + const resetRetryState = yield* Ref.make(false); const state = yield* SubscriptionRef.make( !initialIntent.desired ? availableState(initialIntent, 0) @@ -591,6 +592,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( let pendingRetry = Option.none(); for (;;) { + if (yield* Ref.getAndSet(resetRetryState, false)) { + failureCount = 0; + latestFailure = null; + pendingRetry = Option.none(); + } const currentIntent = yield* Ref.get(intent); if (!currentIntent.desired) { failureCount = 0; @@ -701,7 +707,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( Effect.withSpan("EnvironmentSupervisor.disconnect"), ); - const retryNow = signal({ _tag: "RetryRequested" }).pipe( + const retryNow = Ref.set(resetRetryState, true).pipe( + Effect.andThen(signal({ _tag: "RetryRequested" })), Effect.withSpan("EnvironmentSupervisor.retryNow"), ); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 8013a135896..bfe57a6c0dd 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -56,6 +56,7 @@ export type EnvironmentSubscriptionRpcTag = export type EnvironmentStreamCommandRpcTag = | typeof WS_METHODS.cloudInstallRelayClient + | typeof WS_METHODS.serverUpdateServerWithProgress | typeof WS_METHODS.gitRunStackedAction; export type EnvironmentStreamRpcTag = @@ -80,7 +81,7 @@ export class EnvironmentRpcSubscriptionObserver extends Context.Reference<{ }), }) {} -const isRpcClientError = Schema.is(RpcClientError.RpcClientError); +export const isRpcClientError = Schema.is(RpcClientError.RpcClientError); export type EnvironmentRpcInput = Parameters>[0]; diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index 7584e55d52e..f36087ebf66 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -13,6 +13,7 @@ import { environmentRpcKey, createAtomCommandScheduler, createRuntimeCommand, + scheduleAtomCommandEffect, executeAtomCommand, executeAtomQuery, isAtomCommandInterrupted, @@ -399,6 +400,54 @@ describe("runtime command runner", () => { registry.dispose(); }); + it.effect("releases a shared scheduler lane before the outer command finishes", () => + Effect.gen(function* () { + const handoffStarted = Latch.makeUnsafe(); + const handoffComplete = Latch.makeUnsafe(); + const resumeComplete = Latch.makeUnsafe(); + const runtime = Atom.runtime(Layer.empty); + const scheduler = createAtomCommandScheduler(); + const concurrency = { mode: "serial" as const, key: () => "shared" }; + const updateCommand = createRuntimeCommand(runtime, { + label: "test.update", + execute: (_input: void, registry) => + scheduleAtomCommandEffect( + registry, + scheduler, + concurrency, + undefined, + Effect.sync(() => handoffStarted.openUnsafe()).pipe( + Effect.andThen(handoffComplete.await), + ), + ).pipe(Effect.andThen(resumeComplete.await)), + }); + const configCommand = createRuntimeCommand(runtime, { + label: "test.config", + scheduler, + concurrency, + execute: () => Effect.succeed("configured"), + }); + const registry = AtomRegistry.make(); + + const update = updateCommand.run(registry, undefined); + yield* handoffStarted.await; + const config = configCommand.run(registry, undefined); + handoffComplete.openUnsafe(); + + expect(yield* Effect.promise(() => config)).toMatchObject({ + _tag: "Success", + value: "configured", + waiting: false, + }); + resumeComplete.openUnsafe(); + expect(yield* Effect.promise(() => update)).toMatchObject({ + _tag: "Success", + waiting: false, + }); + registry.dispose(); + }), + ); + it("deduplicates single-flight commands by key", async () => { const latch = Latch.makeUnsafe(); let executions = 0; diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index fb5e9a0ab55..0a00e919c4b 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -253,6 +253,28 @@ export function createAtomCommandScheduler(): AtomCommandScheduler { }; } +/** Runs one effect inside an existing command scheduler lane. */ +export function scheduleAtomCommandEffect( + registry: AtomRegistry.AtomRegistry, + scheduler: AtomCommandScheduler, + concurrency: AtomCommandConcurrency, + input: W, + effect: Effect.Effect, +): Effect.Effect { + return Effect.gen(function* () { + const context = yield* Effect.context(); + const result = yield* Effect.promise((signal) => + scheduler.schedule(registry, concurrency, input, async () => { + const exit = await Effect.runPromiseExitWith(context)(effect, { signal }); + return Exit.isSuccess(exit) + ? AsyncResult.success(exit.value) + : AsyncResult.failure(exit.cause); + }), + ); + return result._tag === "Success" ? result.value : yield* Effect.failCause(result.cause); + }); +} + export async function runAtomCommand( registry: AtomRegistry.AtomRegistry, command: AtomCommand, diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index ddc8813316f..12925a99867 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -6,11 +6,15 @@ import { WS_METHODS, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import { RpcClientError } from "effect/unstable/rpc"; +import * as Socket from "effect/unstable/socket/Socket"; import { AVAILABLE_CONNECTION_STATE, @@ -24,8 +28,12 @@ import type { RpcSession } from "../rpc/session.ts"; import { applyServerConfigProjection, makeEnvironmentServerConfigState, + isLegacyUpdateHandoffLoss, projectServerWelcome, resolveServerConfigValue, + resolveServerUpdateProgressResult, + serverUpdateStateForProgressEvent, + serverUpdateStateForServerVersion, } from "./server.ts"; const CONFIG = { @@ -62,6 +70,109 @@ function session(client: WsRpcProtocolClient): RpcSession { } describe("server state projection", () => { + it("only treats a legacy transport interruption as an unacknowledged handoff", () => { + expect(isLegacyUpdateHandoffLoss(Cause.interrupt(1))).toBe(true); + expect( + isLegacyUpdateHandoffLoss( + Cause.fail( + new RpcClientError.RpcClientError({ + reason: new Socket.SocketCloseError({ code: 1006 }), + }), + ), + ), + ).toBe(true); + expect( + isLegacyUpdateHandoffLoss( + Cause.fail( + new RpcClientError.RpcClientError({ + reason: new Socket.SocketOpenError({ + kind: "Unknown", + cause: new Error("connection refused"), + }), + }), + ), + ), + ).toBe(false); + expect( + isLegacyUpdateHandoffLoss( + Cause.fail( + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "incompatible protocol", + cause: new Error("invalid response"), + }), + }), + ), + ), + ).toBe(false); + expect(isLegacyUpdateHandoffLoss(Cause.fail(new Error("Install failed.")))).toBe(false); + }); + + it.effect("resumes after the progress stream disconnects following completion", () => { + const result = { + targetVersion: "0.0.31", + method: "respawn" as const, + }; + const disconnect = new RpcClientError.RpcClientError({ + reason: new Socket.SocketCloseError({ code: 1006 }), + }); + + return Effect.gen(function* () { + const resumed = yield* resolveServerUpdateProgressResult( + result.targetVersion, + Option.some(result), + Exit.fail(disconnect), + ); + expect(resumed).toEqual(result); + }); + }); + + it("projects streamed update milestones into the shared operation state", () => { + expect( + serverUpdateStateForProgressEvent("0.0.30", "0.0.31", { + type: "progress", + stage: "installing", + }), + ).toEqual({ + status: "running", + stage: "installing", + fromVersion: "0.0.30", + targetVersion: "0.0.31", + }); + expect( + serverUpdateStateForProgressEvent("0.0.30", "0.0.31", { + type: "complete", + result: { targetVersion: "0.0.31", method: "respawn" }, + }), + ).toEqual({ + status: "running", + stage: "resuming", + fromVersion: "0.0.30", + targetVersion: "0.0.31", + }); + }); + + it("keeps active update state and hides stale failures after a version change", () => { + const running = { + status: "running" as const, + stage: "resuming" as const, + fromVersion: "0.0.30", + targetVersion: "0.0.31", + }; + const failed = { + status: "failed" as const, + stage: "installing" as const, + fromVersion: "0.0.30", + targetVersion: "0.0.31", + message: "Install failed.", + }; + + expect(serverUpdateStateForServerVersion(running, "0.0.31")).toBe(running); + expect(serverUpdateStateForServerVersion(failed, "0.0.30")).toBe(failed); + expect(serverUpdateStateForServerVersion(failed, null)).toBe(failed); + expect(serverUpdateStateForServerVersion(failed, "0.0.31")).toEqual({ status: "idle" }); + }); + it("applies every config category to the projected snapshot", () => { const snapshot = applyServerConfigProjection(Option.none(), { version: 1, @@ -100,9 +211,16 @@ describe("server state projection", () => { }); it("prefers an active session config over cache until a live event arrives", () => { - const cached = { ...CONFIG, settings: { source: "cache" } } as unknown as ServerConfig; - const initial = { ...CONFIG, settings: { source: "session" } } as unknown as ServerConfig; - const live = { ...CONFIG, settings: { source: "live" } } as unknown as ServerConfig; + const config = (source: string, serverVersion: string) => + ({ + ...CONFIG, + environment: { serverVersion }, + settings: { source }, + }) as unknown as ServerConfig; + const cached = config("cache", "0.0.29"); + const staleLive = config("stale-live", "0.0.29"); + const initial = config("session", "0.0.30"); + const live = config("live", "0.0.30"); expect( resolveServerConfigValue( @@ -114,6 +232,16 @@ describe("server state projection", () => { initial, ), ).toBe(initial); + expect( + resolveServerConfigValue( + { + config: staleLive, + latestEvent: snapshotEvent(staleLive), + source: "live", + }, + initial, + ), + ).toBe(initial); expect( resolveServerConfigValue( { diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 5f93a3edb6e..edd1893f739 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -3,13 +3,19 @@ import { type ServerConfig, type ServerConfigStreamEvent, type ServerLifecycleWelcomePayload, + type ServerSelfUpdateProgressEvent, + type ServerSelfUpdateResult, WS_METHODS, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -19,14 +25,148 @@ import { createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily, createEnvironmentRpcSubscriptionAtomFamily, + createRuntimeCommand, + scheduleAtomCommandEffect, } from "./runtime.ts"; -import type { EnvironmentRegistry } from "../connection/registry.ts"; +import { EnvironmentRegistry } from "../connection/registry.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts"; +import { + isRpcClientError, + request, + runStream, + subscribe, + type EnvironmentRpcInput, +} from "../rpc/client.ts"; import { followStreamInEnvironment } from "./runtime.ts"; +export type ServerUpdateStage = "downloading" | "installing" | "resuming"; + +export type ServerUpdateState = + | { readonly status: "idle" } + | { + readonly status: "running"; + readonly stage: ServerUpdateStage; + readonly fromVersion: string; + readonly targetVersion: string; + } + | { + readonly status: "failed"; + readonly stage: ServerUpdateStage; + readonly fromVersion: string; + readonly targetVersion: string; + readonly message: string; + }; + +export interface ServerUpdateTarget { + readonly environmentId: EnvironmentId; + readonly input: EnvironmentRpcInput; +} + +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) => + Atom.make(IDLE_SERVER_UPDATE_STATE).pipe( + Atom.withLabel(`environment-data:server:update-state:${environmentId}`), + ), +); + +export class ServerUpdateResumeTimeoutError extends Schema.TaggedErrorClass()( + "ServerUpdateResumeTimeoutError", + { + environmentId: Schema.String, + targetVersion: Schema.String, + }, +) { + override get message(): string { + return `The server did not resume on t3@${this.targetVersion}.`; + } +} + +export class ServerUpdateProgressIncompleteError extends Schema.TaggedErrorClass()( + "ServerUpdateProgressIncompleteError", + { + targetVersion: Schema.String, + }, +) { + override get message(): string { + return `The t3@${this.targetVersion} update ended before the server accepted the restart.`; + } +} + +export function serverUpdateStateForProgressEvent( + fromVersion: string, + targetVersion: string, + event: ServerSelfUpdateProgressEvent, +): Extract { + return { + status: "running", + stage: event.type === "complete" ? "resuming" : event.stage, + fromVersion, + targetVersion, + }; +} + +export function serverUpdateStateForServerVersion( + state: ServerUpdateState, + serverVersion: string | null, +): ServerUpdateState { + return state.status === "idle" || + state.status === "running" || + serverVersion === null || + state.fromVersion === serverVersion + ? state + : IDLE_SERVER_UPDATE_STATE; +} + +function serverUpdateFailureMessage(error: unknown): string { + return error instanceof Error ? error.message : "Server update failed."; +} + +function isRpcSocketError(error: unknown): boolean { + if (!isRpcClientError(error)) { + return false; + } + switch (error.reason._tag) { + case "SocketReadError": + case "SocketWriteError": + case "SocketCloseError": + return true; + default: + return false; + } +} + +export function isLegacyUpdateHandoffLoss(cause: Cause.Cause): boolean { + if (Cause.hasInterruptsOnly(cause)) { + return true; + } + return ( + cause.reasons.length > 0 && + cause.reasons.every((reason) => Cause.isFailReason(reason) && isRpcSocketError(reason.error)) + ); +} + +export function resolveServerUpdateProgressResult( + targetVersion: string, + terminal: Option.Option, + streamExit: Exit.Exit, +): Effect.Effect { + if ( + Option.isSome(terminal) && + (Exit.isSuccess(streamExit) || isLegacyUpdateHandoffLoss(streamExit.cause)) + ) { + return Effect.succeed(terminal.value); + } + if (Exit.isFailure(streamExit)) { + return Effect.failCause(streamExit.cause); + } + return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion })); +} + export interface ServerConfigProjection { readonly config: ServerConfig; readonly latestEvent: ServerConfigStreamEvent; @@ -225,7 +365,13 @@ export function resolveServerConfigValue( projection: ServerConfigProjection | null, initialConfig: ServerConfig | null, ): ServerConfig | null { - if (projection?.source === "live") return projection.config; + if ( + projection?.source === "live" && + (initialConfig === null || + projection.config.environment.serverVersion === initialConfig.environment.serverVersion) + ) { + return projection.config; + } return initialConfig ?? projection?.config ?? null; } @@ -238,6 +384,8 @@ export function createServerEnvironmentAtoms( }, ) { const configScheduler = createAtomCommandScheduler(); + // Updates stay serial end-to-end, but only their handoff phase occupies the config lane. + const updateScheduler = createAtomCommandScheduler(); const configConcurrency = { mode: "serial" as const, key: ({ environmentId }: { readonly environmentId: string }) => environmentId, @@ -271,6 +419,179 @@ export function createServerEnvironmentAtoms( ); }).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`)); }); + const updateStateValueAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => + serverUpdateStateForServerVersion( + get(serverUpdateStateAtom(environmentId)), + get(configValueAtom(environmentId))?.environment.serverVersion ?? null, + ), + ).pipe(Atom.withLabel(`environment-data:server:update-state-value:${environmentId}`)), + ); + const updateStateAtom = (environmentId: EnvironmentId | null) => + environmentId === null ? EMPTY_SERVER_UPDATE_STATE_ATOM : updateStateValueAtom(environmentId); + const updateServer = createRuntimeCommand< + EnvironmentRegistry | EnvironmentCacheStore | R, + E, + ServerUpdateTarget, + ServerSelfUpdateResult, + unknown + >(runtime, { + label: "environment-data:server:update-server", + scheduler: updateScheduler, + concurrency: configConcurrency, + execute: (target, atomRegistry) => { + const stateAtom = serverUpdateStateAtom(target.environmentId); + const targetVersion = target.input.targetVersion; + let fromVersion = + atomRegistry.get(configValueAtom(target.environmentId))?.environment.serverVersion ?? + targetVersion; + let currentStage: ServerUpdateStage = "downloading"; + atomRegistry.set(stateAtom, { + status: "running", + stage: currentStage, + fromVersion, + targetVersion, + }); + + return Effect.gen(function* () { + const environmentRegistry = yield* EnvironmentRegistry; + const result = yield* scheduleAtomCommandEffect( + atomRegistry, + configScheduler, + configConcurrency, + target, + Effect.gen(function* () { + const currentConfig = atomRegistry.get(configValueAtom(target.environmentId)); + fromVersion = currentConfig?.environment.serverVersion ?? targetVersion; + atomRegistry.set(stateAtom, { + status: "running", + stage: currentStage, + fromVersion, + targetVersion, + }); + + const supportsProgress = + currentConfig?.environment.capabilities.serverSelfUpdateProgress === true; + const updateResult: ServerSelfUpdateResult = supportsProgress + ? yield* Effect.gen(function* () { + const terminal = yield* Ref.make>( + Option.none(), + ); + const streamExit = yield* environmentRegistry + .runStream( + target.environmentId, + runStream(WS_METHODS.serverUpdateServerWithProgress, target.input), + ) + .pipe( + Stream.runForEach((event) => + Effect.sync(() => { + currentStage = event.type === "complete" ? "resuming" : event.stage; + atomRegistry.set( + stateAtom, + serverUpdateStateForProgressEvent(fromVersion, targetVersion, event), + ); + }).pipe( + Effect.andThen( + event.type === "complete" + ? Ref.set(terminal, Option.some(event.result)) + : Effect.void, + ), + ), + ), + Effect.exit, + ); + return yield* resolveServerUpdateProgressResult( + targetVersion, + yield* Ref.get(terminal), + streamExit, + ); + }) + : yield* Effect.gen(function* () { + const selfUpdateMethod = currentConfig?.environment.capabilities.serverSelfUpdate; + const exit = yield* environmentRegistry + .run(target.environmentId, request(WS_METHODS.serverUpdateServer, target.input)) + .pipe(Effect.exit); + if (Exit.isSuccess(exit)) { + return exit.value; + } + if ( + (selfUpdateMethod === "boot-service" || selfUpdateMethod === "respawn") && + isLegacyUpdateHandoffLoss(exit.cause) + ) { + // Older servers can tear down the transport before their + // unary acknowledgement arrives. Treat only that transport + // loss as a handoff, then prove it by waiting for target ready. + return { targetVersion, method: selfUpdateMethod }; + } + return yield* Effect.failCause(exit.cause); + }); + + currentStage = "resuming"; + atomRegistry.set(stateAtom, { + status: "running", + stage: currentStage, + fromVersion, + targetVersion, + }); + return updateResult; + }), + ); + + // The update restart is intentional. As soon as the supervisor sees + // that first failed connection, discard any prior backoff debt and + // retry immediately instead of carrying an old 16-second delay. + yield* environmentRegistry.stateChanges(target.environmentId).pipe( + Stream.filter((state) => state.phase === "backoff"), + Stream.take(1), + Stream.runDrain, + Effect.andThen(environmentRegistry.retryNow(target.environmentId)), + Effect.timeoutOption(Duration.seconds(30)), + Effect.ignore, + Effect.forkChild, + ); + + const resumed = yield* environmentRegistry + .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {})) + .pipe( + Stream.filter( + (event) => + event.type === "ready" && event.payload.environment.serverVersion === targetVersion, + ), + Stream.runHead, + Effect.timeoutOption(Duration.seconds(120)), + Effect.map(Option.flatten), + ); + if (Option.isNone(resumed)) { + return yield* new ServerUpdateResumeTimeoutError({ + environmentId: target.environmentId, + targetVersion, + }); + } + + atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE); + return result; + }).pipe( + Effect.onExit((exit) => + Effect.sync(() => { + if (Exit.isSuccess(exit)) { + return; + } + if (Cause.hasInterruptsOnly(exit.cause)) { + atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE); + return; + } + atomRegistry.set(stateAtom, { + status: "failed", + stage: currentStage, + fromVersion, + targetVersion, + message: serverUpdateFailureMessage(Cause.squash(exit.cause)), + }); + }), + ), + ); + }, + }); const settingsValueAtom = Atom.family((environmentId: EnvironmentId) => Atom.make((get) => get(configValueAtom(environmentId))?.settings ?? null).pipe( Atom.withLabel(`environment-data:server:settings:${environmentId}`), @@ -284,6 +605,7 @@ export function createServerEnvironmentAtoms( return { configValueAtom, + updateStateAtom, settingsValueAtom, providersValueAtom, traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, { @@ -331,12 +653,7 @@ export function createServerEnvironmentAtoms( scheduler: configScheduler, concurrency: configConcurrency, }), - updateServer: createEnvironmentRpcCommand(runtime, { - label: "environment-data:server:update-server", - tag: WS_METHODS.serverUpdateServer, - scheduler: configScheduler, - concurrency: configConcurrency, - }), + updateServer, upsertKeybinding: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:upsert-keybinding", tag: WS_METHODS.serverUpsertKeybinding, diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 7f4b6c16541..7b1769fdfc0 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -51,6 +51,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ servers that must be relaunched manually (dev checkouts, Windows foreground runs, pre-update servers). */ serverSelfUpdate: Schema.optionalKey(ServerSelfUpdateCapability), + /** Server can stream self-update progress before acknowledging the + restart. Clients fall back to server.updateServer when absent. */ + serverSelfUpdateProgress: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 0701e15a668..d1a5f2504c7 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -127,6 +127,7 @@ import { ServerProviderUpdatedPayload, ServerSelfUpdateError, ServerSelfUpdateInput, + ServerSelfUpdateProgressEvent, ServerSelfUpdateResult, ServerTraceDiagnosticsResult, ServerProcessDiagnosticsResult, @@ -218,6 +219,7 @@ export const WS_METHODS = { serverRefreshProviders: "server.refreshProviders", serverUpdateProvider: "server.updateProvider", serverUpdateServer: "server.updateServer", + serverUpdateServerWithProgress: "server.updateServerWithProgress", serverUpsertKeybinding: "server.upsertKeybinding", serverRemoveKeybinding: "server.removeKeybinding", serverGetSettings: "server.getSettings", @@ -305,6 +307,16 @@ export const WsServerUpdateServerRpc = Rpc.make(WS_METHODS.serverUpdateServer, { error: Schema.Union([ServerSelfUpdateError, EnvironmentAuthorizationError]), }); +export const WsServerUpdateServerWithProgressRpc = Rpc.make( + WS_METHODS.serverUpdateServerWithProgress, + { + payload: ServerSelfUpdateInput, + success: ServerSelfUpdateProgressEvent, + error: Schema.Union([ServerSelfUpdateError, EnvironmentAuthorizationError]), + stream: true, + }, +); + export const WsServerGetSettingsRpc = Rpc.make(WS_METHODS.serverGetSettings, { payload: Schema.Struct({}), success: ServerSettings, @@ -759,6 +771,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerRefreshProvidersRpc, WsServerUpdateProviderRpc, WsServerUpdateServerRpc, + WsServerUpdateServerWithProgressRpc, WsServerUpsertKeybindingRpc, WsServerRemoveKeybindingRpc, WsServerGetSettingsRpc, diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index 8e42c938ca4..e083523bbdf 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -592,6 +592,21 @@ export const ServerSelfUpdateResult = Schema.Struct({ }); export type ServerSelfUpdateResult = typeof ServerSelfUpdateResult.Type; +export const ServerSelfUpdateProgressStage = Schema.Literals(["downloading", "installing"]); +export type ServerSelfUpdateProgressStage = typeof ServerSelfUpdateProgressStage.Type; + +export const ServerSelfUpdateProgressEvent = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("progress"), + stage: ServerSelfUpdateProgressStage, + }), + Schema.Struct({ + type: Schema.Literal("complete"), + result: ServerSelfUpdateResult, + }), +]); +export type ServerSelfUpdateProgressEvent = typeof ServerSelfUpdateProgressEvent.Type; + export class ServerSelfUpdateError extends Schema.TaggedErrorClass()( "ServerSelfUpdateError", {