diff --git a/apps/desktop/src/app/DesktopConnectionCatalogStore.ts b/apps/desktop/src/app/DesktopConnectionCatalogStore.ts index 5ec2edb595ff..fa0252059ef4 100644 --- a/apps/desktop/src/app/DesktopConnectionCatalogStore.ts +++ b/apps/desktop/src/app/DesktopConnectionCatalogStore.ts @@ -369,6 +369,7 @@ const migrateSavedEnvironmentRecords = Effect.fn( return { schemaVersion: 1, targets, + routes: targets, profiles, credentials, remoteDpopTokens: [], diff --git a/apps/mobile/src/connection/storage.ts b/apps/mobile/src/connection/storage.ts index e844181673df..fe5489e369d4 100644 --- a/apps/mobile/src/connection/storage.ts +++ b/apps/mobile/src/connection/storage.ts @@ -3,9 +3,12 @@ import { ConnectionRegistrationStore, ConnectionTargetStore, registerConnectionInCatalog, + removeConnectionRouteFromCatalog, removeConnectionFromCatalog, removeCatalogValue, replaceCatalogValue, + selectConnectionRouteInCatalog, + connectionRoutes, } from "@t3tools/client-runtime/platform"; import { TokenStore } from "@t3tools/client-runtime/authorization"; import { @@ -20,7 +23,13 @@ import * as Option from "effect/Option"; import * as CatalogStore from "./catalog-store"; function targetPersistenceError( - operation: "list-targets" | "register-connection" | "remove-connection", + operation: + | "list-targets" + | "list-routes" + | "register-connection" + | "select-connection-route" + | "remove-connection-route" + | "remove-connection", error: ConnectionTransientError, ) { return new ConnectionPersistenceError({ @@ -38,12 +47,28 @@ export const connectionStorageLayer = Layer.effectContext( Effect.map((document) => document.targets), Effect.mapError((error) => targetPersistenceError("list-targets", error)), ), + listRoutes: catalog.read.pipe( + Effect.map(connectionRoutes), + Effect.mapError((error) => targetPersistenceError("list-routes", error)), + ), }); const registrationStore = ConnectionRegistrationStore.of({ register: (registration) => catalog .update((document) => registerConnectionInCatalog(document, registration)) .pipe(Effect.mapError((error) => targetPersistenceError("register-connection", error))), + select: (target) => + catalog + .update((document) => selectConnectionRouteInCatalog(document, target)) + .pipe( + Effect.mapError((error) => targetPersistenceError("select-connection-route", error)), + ), + removeRoute: (target, fallback) => + catalog + .update((document) => removeConnectionRouteFromCatalog(document, target, fallback)) + .pipe( + Effect.mapError((error) => targetPersistenceError("remove-connection-route", error)), + ), remove: (target) => catalog .update((document) => removeConnectionFromCatalog(document, target)) diff --git a/apps/web/src/components/settings/ConnectionsSettings.tsx b/apps/web/src/components/settings/ConnectionsSettings.tsx index 18d1b0f1c924..c3bbd75cf116 100644 --- a/apps/web/src/components/settings/ConnectionsSettings.tsx +++ b/apps/web/src/components/settings/ConnectionsSettings.tsx @@ -28,7 +28,11 @@ import { type DesktopWslState, type EnvironmentId, } from "@t3tools/contracts"; -import { connectionStatusText } from "@t3tools/client-runtime/connection"; +import { + type ConnectionCatalogEntry, + connectionRouteId, + connectionStatusText, +} from "@t3tools/client-runtime/connection"; import { isAtomCommandInterrupted, squashAtomCommandFailure, @@ -1350,16 +1354,37 @@ function NetworkAccessDescription({ type SavedBackendListRowProps = { environment: EnvironmentPresentation; + routes: ReadonlyArray; removingEnvironmentId: EnvironmentId | null; + switchingRouteEnvironmentIds: ReadonlySet; onConnect: (environmentId: EnvironmentId) => void; onRemove: (environmentId: EnvironmentId) => void; + onSelectRoute: (environmentId: EnvironmentId, routeId: string) => void; }; +function connectionRouteLabel(entry: ConnectionCatalogEntry): string { + switch (entry.target._tag) { + case "RelayConnectionTarget": + return "T3 Connect"; + case "SshConnectionTarget": + return Option.isSome(entry.profile) && entry.profile.value._tag === "SshConnectionProfile" + ? `SSH ${formatDesktopSshTarget(entry.profile.value.target)}` + : "SSH"; + case "BearerConnectionTarget": + return "Direct"; + case "PrimaryConnectionTarget": + return "Local"; + } +} + function SavedBackendListRow({ environment, + routes, removingEnvironmentId, + switchingRouteEnvironmentIds, onConnect, onRemove, + onSelectRoute, }: SavedBackendListRowProps) { const environmentId = environment.environmentId; const connectionState = environment.connection.phase; @@ -1478,6 +1503,43 @@ function SavedBackendListRow({ ) : null}
+ {routes.length > 1 ? ( + + ) : null} {versionMismatch && (serverUpdateState.status === "idle" || serverUpdateState.status === "failed") ? ( (null); + const [switchingRouteEnvironmentIds, setSwitchingRouteEnvironmentIds] = useState< + ReadonlySet + >(new Set()); const [connectingSshHostAlias, setConnectingSshHostAlias] = useState(null); const [desktopServerExposureMutationError, setDesktopServerExposureMutationError] = useState< @@ -2265,6 +2334,33 @@ export function ConnectionsSettings() { [retryEnvironment], ); + const handleSelectSavedBackendRoute = useCallback( + async (environmentId: EnvironmentId, routeId: string) => { + setSwitchingRouteEnvironmentIds((current) => new Set(current).add(environmentId)); + setSavedBackendError(null); + const result = await selectEnvironmentRoute({ environmentId, routeId }); + setSwitchingRouteEnvironmentIds((current) => { + const next = new Set(current); + next.delete(environmentId); + return next; + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + const message = + error instanceof Error ? error.message : "Failed to switch connection method."; + setSavedBackendError(message); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Could not switch connection method", + description: message, + }), + ); + } + }, + [selectEnvironmentRoute], + ); + const handleRemoveSavedBackend = useCallback( async (environmentId: EnvironmentId) => { setRemovingSavedEnvironmentId(environmentId); @@ -3448,9 +3544,12 @@ export function ConnectionsSettings() { ))} ; + }, ): DesktopBridge { return { - ensureSshEnvironment: async (target: DesktopSshEnvironmentTarget) => { + ensureSshEnvironment: async ( + target: DesktopSshEnvironmentTarget, + ensureOptions?: { readonly issuePairingToken?: boolean }, + ) => { calls.push("ensure"); + options?.ensurePairingOptions?.push(ensureOptions?.issuePairingToken ?? null); return { target, httpBaseUrl: "http://127.0.0.1:3201/", @@ -95,6 +103,41 @@ describe("desktop SSH pairing", () => { expect(calls).toEqual(["ensure", "descriptor"]); }), ); + + it.effect("reuses a cached bearer without requesting or bootstrapping a pairing token", () => + Effect.gen(function* () { + const calls: string[] = []; + const ensurePairingOptions: Array = []; + + const prepared = yield* prepareDesktopSshEnvironment( + makeBridge(calls, { ensurePairingOptions }), + { + target: TARGET, + bearerToken: "cached-bearer", + }, + ); + + expect(prepared.bearerToken).toBe("cached-bearer"); + expect(calls).toEqual(["ensure"]); + expect(ensurePairingOptions).toEqual([false]); + }), + ); + + it.effect("requests and bootstraps a pairing token when no bearer is cached", () => + Effect.gen(function* () { + const calls: string[] = []; + const ensurePairingOptions: Array = []; + + const prepared = yield* prepareDesktopSshEnvironment( + makeBridge(calls, { ensurePairingOptions }), + { target: TARGET }, + ); + + expect(prepared.bearerToken).toBe("bearer-token"); + expect(calls).toEqual(["ensure", "token"]); + expect(ensurePairingOptions).toEqual([true]); + }), + ); }); describe("desktop-local bearer cache", () => { diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index daead0bc6308..cb82d300835f 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -173,6 +173,44 @@ export const provisionDesktopSshEnvironment = Effect.fn( }; }); +export const prepareDesktopSshEnvironment = Effect.fn("web.connectionPlatform.ssh.prepareDesktop")( + function* ( + bridge: DesktopBridge, + input: { + readonly target: DesktopSshEnvironmentTarget; + readonly bearerToken?: string; + }, + ) { + const bootstrap = yield* Effect.tryPromise({ + try: () => + bridge.ensureSshEnvironment(input.target, { + issuePairingToken: input.bearerToken === undefined, + }), + catch: sshPreparationError, + }); + if (input.bearerToken !== undefined) { + return { + bootstrap, + bearerToken: input.bearerToken, + }; + } + if (bootstrap.pairingToken === null) { + return yield* new ConnectionBlockedError({ + reason: "authentication", + detail: "The SSH environment did not issue a pairing credential.", + }); + } + const access = yield* Effect.tryPromise({ + try: () => bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, bootstrap.pairingToken!), + catch: sshPreparationError, + }); + return { + bootstrap, + bearerToken: access.access_token, + }; + }, +); + const capabilitiesLayer = Layer.effectContext( Effect.sync(() => { const presentation = ClientPresentation.of({ @@ -238,28 +276,7 @@ const capabilitiesLayer = Layer.effectContext( detail: "SSH environments are only available in the desktop app.", }); } - const bootstrap = yield* Effect.tryPromise({ - try: () => - bridge.ensureSshEnvironment(input.target, { - issuePairingToken: true, - }), - catch: sshPreparationError, - }); - if (bootstrap.pairingToken === null) { - return yield* new ConnectionBlockedError({ - reason: "authentication", - detail: "The SSH environment did not issue a pairing credential.", - }); - } - const access = yield* Effect.tryPromise({ - try: () => - bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, bootstrap.pairingToken!), - catch: sshPreparationError, - }); - return { - bootstrap, - bearerToken: access.access_token, - }; + return yield* prepareDesktopSshEnvironment(bridge, input); }), disconnect: Effect.fn("web.connectionPlatform.ssh.disconnect")(function* (target) { const bridge = window.desktopBridge; diff --git a/apps/web/src/connection/storage.test.ts b/apps/web/src/connection/storage.test.ts index 6d503387bb64..89e792b2471e 100644 --- a/apps/web/src/connection/storage.test.ts +++ b/apps/web/src/connection/storage.test.ts @@ -10,6 +10,7 @@ import { makeCatalogBackend, makeCatalogStore } from "./storage"; const emptyCatalog = { schemaVersion: 1, targets: [], + routes: [], profiles: [], credentials: [], remoteDpopTokens: [], diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 4ae476c1d11f..eee54b0adda3 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -8,8 +8,11 @@ import { EnvironmentCacheStore, registerConnectionInCatalog, removeCatalogValue, + removeConnectionRouteFromCatalog, removeConnectionFromCatalog, replaceCatalogValue, + selectConnectionRouteInCatalog, + connectionRoutes, } from "@t3tools/client-runtime/platform"; import { TokenStore } from "@t3tools/client-runtime/authorization"; import { @@ -97,7 +100,10 @@ function catalogError(operation: string, cause: unknown) { function persistenceError( operation: | "list-targets" + | "list-routes" | "register-connection" + | "select-connection-route" + | "remove-connection-route" | "remove-connection" | "load-shell" | "save-shell" @@ -375,12 +381,24 @@ export const connectionStorageLayer = Layer.effectContext( Effect.map((document) => document.targets), Effect.mapError((cause) => persistenceError("list-targets", cause)), ), + listRoutes: catalog.read.pipe( + Effect.map(connectionRoutes), + Effect.mapError((cause) => persistenceError("list-routes", cause)), + ), }); const registrationStore = ConnectionRegistrationStore.of({ register: (registration) => catalog .update((document) => registerConnectionInCatalog(document, registration)) .pipe(Effect.mapError((cause) => persistenceError("register-connection", cause))), + select: (target) => + catalog + .update((document) => selectConnectionRouteInCatalog(document, target)) + .pipe(Effect.mapError((cause) => persistenceError("select-connection-route", cause))), + removeRoute: (target, fallback) => + catalog + .update((document) => removeConnectionRouteFromCatalog(document, target, fallback)) + .pipe(Effect.mapError((cause) => persistenceError("remove-connection-route", cause))), remove: (target) => catalog .update((document) => removeConnectionFromCatalog(document, target)) diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 46fe0c82716a..6f9cf052421e 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -35,6 +35,14 @@ scope when it changed. `createServiceScope` builds an `EnvironmentSupervisor` bound to a closeable scope and connects it; `run` and `runStream` execute caller effects with that supervisor provided. +Persisted environments may have multiple routes but exactly one selected target +per client. `EnvironmentRegistry.routes` publishes the retained route entries. +`selectRoute` first calls `ConnectionDriver.prepare` while the selected +supervisor remains live, so authentication and stable environment identity are +verified before persistence changes. It then swaps the selected entry and +supervisor. Selection never falls back automatically; a failed preparation +leaves the existing session and persisted selection untouched. + `EnvironmentSupervisor` owns desired state, retry scheduling, and the active session scope. React components do not create connections, transports, retry loops, or RPC clients. diff --git a/docs/internals/remote.md b/docs/internals/remote.md index afce95f725bc..d9301e130a0a 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -57,8 +57,10 @@ control plane or a copy of session state. | `RelayConnectionTarget` | Managed T3 Connect relay tunnels. | | `SshConnectionTarget` | Desktop-managed SSH environments. | -Bearer, relay, and SSH are persisted; primary is platform-managed. Note that Tailscale is not a -separate target kind. A Tailscale URL is paired through the ordinary bearer path in +Bearer, relay, and SSH routes are persisted; primary is platform-managed. The catalog keeps one +selected target per environment plus every known route for that stable environment ID. Registering +an SSH route for an environment already reached through the relay retains both routes instead of +replacing the relay registration. Note that Tailscale is not a separate target kind. A Tailscale URL is paired through the ordinary bearer path in [`onboarding.ts`][onboarding] (`preparePairingRegistration`), which accepts either a pairing URL or a host plus pairing code. Tailscale is an endpoint provider and transport, not a distinct runtime concept. @@ -166,6 +168,17 @@ server, opens a local tunnel, checks HTTP readiness, optionally issues a remote returns local HTTP/WS endpoints. Disconnect closes the tunnel and stops the remote server if the launcher started it; a server that was already running (marked `external`) is left running. +Remote discovery checks the user-level `t3code.service` process first, then the SSH environment and +same-user `/proc` metadata for `T3CODE_HOME` values. It selects the first authoritative runtime whose +PID is live and whose recorded origin is loopback, then waits for that runtime instead of falling +through to a lower-priority server when it is temporarily busy. Once tunneled, the client fetches the +environment descriptor and verifies the stable environment ID before changing routes. The discovered +base directory is saved for `t3 pair --base-dir`, so a required pairing credential lands in the +database the reused server actually reads. Normal reconnects and route switches reuse the stored +bearer credential rather than writing a new pairing record. This is what prevents an SSH access path +from creating a second logical environment beside an already-running T3 Connect service whose private +service environment is not inherited by SSH. + The desktop main process owns this because it can spawn SSH, manage prompts, write launch scripts, and clean up forwards. The renderer connects through the forwarded URL like any other environment and needs no SSH-specific RPC path. @@ -175,6 +188,13 @@ failure includes launcher output where available, forwarded-port failure leaves disconnected rather than falling back to an unrelated endpoint, and reconnect restores the SSH bridge before reconnecting the WebSocket client. +Route selection is manual and device-local. `EnvironmentRegistry.selectRoute` prepares and +authenticates the candidate while the current supervisor remains active, including verifying its +environment ID. Only then does it persist the selected target and replace the supervisor. There is +one active RPC session per environment on a client; an inactive SSH tunnel may remain warm for a +later manual switch, but it owns no subscriptions. A failed candidate leaves the selected route and +active session unchanged. Mobile shares the route-capable catalog model but exposes no SSH UI. + ## Launch methods Launch answers a different question: how does a T3 server come to exist on the target machine? Keep diff --git a/docs/user/remote-access.md b/docs/user/remote-access.md index 5993fca5b352..674c466ebf11 100644 --- a/docs/user/remote-access.md +++ b/docs/user/remote-access.md @@ -136,10 +136,27 @@ After setup, the renderer connects to a local forwarded HTTP/WebSocket endpoint. SSH launch is a desktop feature because it needs local process and SSH access. Once the environment is paired and saved, it uses the same environment list and connection model as direct LAN, Tailscale, HTTPS, or future tunnel-backed environments. +If the same server is already saved through T3 Connect, adding it through SSH does not create a +second environment. The desktop keeps both access methods under the server's stable environment +identity. Choose **T3 Connect** or **SSH** from the environment's connection-method menu. The choice +is manual and applies only to that desktop; T3 Code does not silently fall back to the other method. +Before switching, the desktop prepares the selected route and verifies that it reaches the same +environment, then hands the active session over. If preparation fails, the current connection stays +active. + +All clients still talk to the same T3 server, regardless of their route. A thread created by the +desktop over SSH is therefore available to mobile over T3 Connect, and server events continue to fan +out to every connected client. + #### SSH Launch Troubleshooting The desktop SSH launcher connects with a non-interactive `sh` session, writes a small launcher script under `~/.t3/ssh-launch//`, starts or reuses a remote T3 server, and forwards the remote loopback port back to your desktop. +Before starting another server, the launcher inspects the remote T3 service and running-process +metadata, then validates the recorded server through its environment descriptor. This lets SSH reuse +a server started by T3 Connect even when that service uses a private `T3CODE_HOME` that the SSH login +shell does not inherit; the forwarded remote port is the port reported by that server. + The remote host must have a compatible Node.js runtime. T3 Code uses the server package's `engines.node` requirement: ```text @@ -165,7 +182,10 @@ nvm alias default 24 With mise, asdf, fnm, or nodenv, make sure the tool's shim directory is installed and resolves to a Node version satisfying the range above without an interactive shell. -If reconnecting after an app update fails, retry the SSH launch once. The launcher now compares its generated runner script, stops stale launcher-managed remote servers, clears the SSH launch PID/port state, and starts a fresh remote server. You should not normally need to delete `~/.t3/ssh-launch` or kill `t3` processes manually. +If reconnecting after an app update fails, use **Connect** to retry. The launcher compares its +generated runner script, clears stale launcher state, and starts a fresh server only when it cannot +discover or reuse a live one. You should not normally need to delete `~/.t3/ssh-launch` or kill `t3` +processes manually. ## Updating a Remote Server diff --git a/packages/client-runtime/src/connection/catalog.ts b/packages/client-runtime/src/connection/catalog.ts index 8df3b2541d18..19ebdd2eedc2 100644 --- a/packages/client-runtime/src/connection/catalog.ts +++ b/packages/client-runtime/src/connection/catalog.ts @@ -79,6 +79,7 @@ export class SshConnectionRegistration extends Schema.TaggedClass Effect.Effect; readonly connect: ( entry: ConnectionCatalogEntry, reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect, @@ -39,6 +42,7 @@ export class ConnectionDriver extends Context.Service< export const make = Effect.gen(function* () { const resolver = yield* ConnectionResolver.ConnectionResolver; const sessions = yield* RpcSession.RpcSessionFactory; + const prepare = resolver.prepare; const connect = Effect.fn("ConnectionDriver.connect")(function* ( entry: ConnectionCatalogEntry, @@ -50,7 +54,7 @@ export const make = Effect.gen(function* () { "connection.target.kind": target._tag, }); yield* reportProgress({ stage: "preparing" }); - const prepared = yield* resolver.prepare(entry); + const prepared = yield* prepare(entry); yield* reportProgress({ stage: "opening", prepared }); const session = yield* sessions.connect(prepared); yield* reportProgress({ stage: "synchronizing", prepared }); @@ -58,7 +62,7 @@ export const make = Effect.gen(function* () { return { prepared, session } satisfies EnvironmentConnectionLease; }); - return ConnectionDriver.of({ connect }); + return ConnectionDriver.of({ prepare, connect }); }); export const layer = Layer.effect(ConnectionDriver, make); diff --git a/packages/client-runtime/src/connection/index.ts b/packages/client-runtime/src/connection/index.ts index 53a041bbf307..6943a6746980 100644 --- a/packages/client-runtime/src/connection/index.ts +++ b/packages/client-runtime/src/connection/index.ts @@ -24,9 +24,11 @@ export { export * from "./presentation.ts"; export * as ProfileStore from "./profileStore.ts"; export { + ConnectionRouteNotRegisteredError, EnvironmentNotRegisteredError, EnvironmentRegistry, PlatformEnvironmentRemovalError, + PlatformEnvironmentRouteSelectionError, } from "./registry.ts"; export { ConnectionResolver } from "./resolver.ts"; export { EnvironmentSupervisor, type EnvironmentSupervisorOptions } from "./supervisor.ts"; diff --git a/packages/client-runtime/src/connection/model.ts b/packages/client-runtime/src/connection/model.ts index fbcb302ed135..722b5cea2cd1 100644 --- a/packages/client-runtime/src/connection/model.ts +++ b/packages/client-runtime/src/connection/model.ts @@ -55,6 +55,18 @@ export type PersistedConnectionTarget = typeof PersistedConnectionTarget.Type; export type ConnectionTargetKind = ConnectionTarget["_tag"]; +export function connectionRouteId(target: ConnectionTarget): string { + switch (target._tag) { + case "PrimaryConnectionTarget": + return `primary:${target.environmentId}`; + case "RelayConnectionTarget": + return `relay:${target.environmentId}`; + case "BearerConnectionTarget": + case "SshConnectionTarget": + return target.connectionId; + } +} + export type NetworkStatus = "unknown" | "offline" | "online"; export const ConnectionTransientReason = Schema.Literals([ diff --git a/packages/client-runtime/src/connection/onboarding.test.ts b/packages/client-runtime/src/connection/onboarding.test.ts index 9bee0dad6fb0..0042a22d2a34 100644 --- a/packages/client-runtime/src/connection/onboarding.test.ts +++ b/packages/client-runtime/src/connection/onboarding.test.ts @@ -251,6 +251,10 @@ describe("connection onboarding", () => { connectionId: "ssh:environment-ssh", target, }, + credential: { + _tag: "BearerConnectionCredential", + token: "bearer-token", + }, }); }), ); diff --git a/packages/client-runtime/src/connection/onboarding.ts b/packages/client-runtime/src/connection/onboarding.ts index e76bcd50a2cc..0809ac8b8538 100644 --- a/packages/client-runtime/src/connection/onboarding.ts +++ b/packages/client-runtime/src/connection/onboarding.ts @@ -230,6 +230,9 @@ export const prepareSshRegistration = Effect.fn( label, target: provisioned.bootstrap.target, }), + credential: new BearerConnectionCredential({ + token: provisioned.bearerToken, + }), }); }); diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 9354db9c998a..b13267d4a49b 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -39,6 +39,7 @@ import { type ConnectionTarget, type PreparedConnection, type SupervisorConnectionState, + connectionRouteId, } from "./model.ts"; import * as Persistence from "../platform/persistence.ts"; import * as ConnectionProfileStore from "./profileStore.ts"; @@ -105,6 +106,10 @@ const SSH_CONNECTION = new SshConnectionTarget({ label: "SSH environment", connectionId: "ssh-connection", }); +const SSH_RELAY_TARGET = new RelayConnectionTarget({ + environmentId: SSH_CONNECTION.environmentId, + label: SSH_CONNECTION.label, +}); const SSH_PROFILE = new SshConnectionProfile({ connectionId: SSH_CONNECTION.connectionId, environmentId: SSH_CONNECTION.environmentId, @@ -128,6 +133,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( initialProfiles: ReadonlyArray = [], initialCredentials: ReadonlyArray = [], options?: { + readonly initialRoutes?: ReadonlyArray; + readonly prepareRoute?: ( + target: ConnectionTarget, + ) => Effect.Effect; readonly beforeSessionConnect?: (environmentId: EnvironmentId) => Effect.Effect; readonly beforeRegistrationRegister?: ( registration: ConnectionRegistration, @@ -140,6 +149,14 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( const storedTargets = yield* Ref.make( new Map(initialTargets.map((target) => [target.environmentId, target])), ); + const storedRoutes = yield* Ref.make( + new Map( + (options?.initialRoutes ?? initialTargets).map((target) => [ + connectionRouteId(target), + target, + ]), + ), + ); const shellCache = yield* Ref.make(new Map([[TARGET.environmentId, CACHED_SNAPSHOT]])); const cacheClears = yield* Ref.make>([]); const ownedDataClears = yield* Ref.make>([]); @@ -173,6 +190,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( const targetStore = Persistence.ConnectionTargetStore.of({ list: Ref.get(storedTargets).pipe(Effect.map((targets) => [...targets.values()])), + listRoutes: Ref.get(storedRoutes).pipe(Effect.map((routes) => [...routes.values()])), }); const registrationStore = Persistence.ConnectionRegistrationStore.of({ register: (registration) => @@ -183,6 +201,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( next.set(registration.target.environmentId, registration.target); return next; }); + yield* Ref.update(storedRoutes, (current) => { + const next = new Map(current); + next.set(connectionRouteId(registration.target), registration.target); + return next; + }); switch (registration._tag) { case "RelayConnectionRegistration": return; @@ -204,28 +227,83 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( next.set(registration.profile.connectionId, registration.profile); return next; }); + if (registration.credential !== undefined) { + yield* Ref.update(storedCredentials, (current) => { + const next = new Map(current); + next.set(registration.target.connectionId, registration.credential!); + return next; + }); + } } }), - remove: (target) => + select: (target) => + Ref.update(storedTargets, (current) => { + const next = new Map(current); + next.set(target.environmentId, target); + return next; + }), + removeRoute: (target, fallback) => Effect.gen(function* () { yield* options?.beforeRegistrationRemove?.(target) ?? Effect.void; yield* Ref.update(storedTargets, (current) => { const next = new Map(current); - next.delete(target.environmentId); + if ( + connectionRouteId(next.get(target.environmentId) ?? fallback) === + connectionRouteId(target) + ) { + next.set(target.environmentId, fallback); + } return next; }); - if (target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget") { - yield* Ref.update(storedProfiles, (current) => { - const next = new Map(current); - next.delete(target.connectionId); - return next; - }); - yield* Ref.update(storedCredentials, (current) => { + yield* Ref.update(storedRoutes, (current) => { + const next = new Map(current); + next.delete(connectionRouteId(target)); + return next; + }); + if (target._tag === "RelayConnectionTarget") { + yield* Ref.update(storedRemoteTokens, (current) => { const next = new Map(current); - next.delete(target.connectionId); + next.delete(target.environmentId); return next; }); } + }), + remove: (target) => + Effect.gen(function* () { + yield* options?.beforeRegistrationRemove?.(target) ?? Effect.void; + const removedConnectionIds = new Set( + [...(yield* Ref.get(storedRoutes)).values()] + .filter((route) => route.environmentId === target.environmentId) + .flatMap((route) => + route._tag === "BearerConnectionTarget" || route._tag === "SshConnectionTarget" + ? [route.connectionId] + : [], + ), + ); + yield* Ref.update(storedTargets, (current) => { + const next = new Map(current); + next.delete(target.environmentId); + return next; + }); + yield* Ref.update(storedRoutes, (current) => { + const next = new Map(current); + for (const [routeId, route] of next) { + if (route.environmentId === target.environmentId) { + next.delete(routeId); + } + } + return next; + }); + yield* Ref.update(storedProfiles, (current) => { + const next = new Map(current); + for (const connectionId of removedConnectionIds) next.delete(connectionId); + return next; + }); + yield* Ref.update(storedCredentials, (current) => { + const next = new Map(current); + for (const connectionId of removedConnectionIds) next.delete(connectionId); + return next; + }); yield* Ref.update(storedRemoteTokens, (current) => { const next = new Map(current); next.delete(target.environmentId); @@ -334,6 +412,14 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( disconnect: (target) => Ref.update(disconnectedSshTargets, (current) => [...current, target]), }); const driver = ConnectionDriver.ConnectionDriver.of({ + prepare: (entry) => + options?.prepareRoute?.(entry.target) ?? + Effect.succeed({ + ...PREPARED, + environmentId: entry.target.environmentId, + label: entry.target.label, + target: entry.target, + }), connect: (entry, reportProgress) => Effect.gen(function* () { const target = entry.target; @@ -389,6 +475,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( return { layer, storedTargets, + storedRoutes, shellCache, cacheClears, ownedDataClears, @@ -597,6 +684,95 @@ describe("EnvironmentRegistry", () => { }), ); + it.effect("prepares and verifies a route before replacing the active session", () => + Effect.gen(function* () { + const prepareStarted = yield* Deferred.make(); + const releasePreparation = yield* Deferred.make(); + const harness = yield* makeHarness([SSH_RELAY_TARGET], [SSH_PROFILE], [], { + initialRoutes: [SSH_RELAY_TARGET, SSH_CONNECTION], + prepareRoute: (target) => + Deferred.succeed(prepareStarted, undefined).pipe( + Effect.andThen(Deferred.await(releasePreparation)), + Effect.as({ + ...PREPARED, + environmentId: target.environmentId, + label: target.label, + target, + }), + ), + }); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.start; + yield* awaitConnectionState( + registry, + SSH_RELAY_TARGET.environmentId, + (state) => state.phase === "connected", + ); + + const selection = yield* Effect.forkChild( + registry.selectRoute(SSH_RELAY_TARGET.environmentId, connectionRouteId(SSH_CONNECTION)), + ); + yield* Deferred.await(prepareStarted); + expect(yield* Ref.get(harness.releasedSessions)).toBe(0); + yield* Deferred.succeed(releasePreparation, undefined); + yield* Fiber.join(selection); + yield* awaitConnectionState( + registry, + SSH_CONNECTION.environmentId, + (state) => state.phase === "connected", + ); + + expect((yield* Ref.get(harness.storedTargets)).get(SSH_CONNECTION.environmentId)).toEqual( + SSH_CONNECTION, + ); + expect( + (yield* SubscriptionRef.get(registry.entries)).get(SSH_CONNECTION.environmentId)?.target, + ).toEqual(SSH_CONNECTION); + expect(yield* Ref.get(harness.releasedSessions)).toBe(1); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + + it.effect("keeps the active route when candidate preparation fails", () => + Effect.gen(function* () { + const harness = yield* makeHarness([SSH_RELAY_TARGET], [SSH_PROFILE], [], { + initialRoutes: [SSH_RELAY_TARGET, SSH_CONNECTION], + prepareRoute: () => + Effect.fail( + new ConnectionTransientError({ + reason: "remote-unavailable", + detail: "SSH is unavailable.", + }), + ), + }); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.start; + yield* awaitConnectionState( + registry, + SSH_RELAY_TARGET.environmentId, + (state) => state.phase === "connected", + ); + + const failure = yield* Effect.flip( + registry.selectRoute(SSH_RELAY_TARGET.environmentId, connectionRouteId(SSH_CONNECTION)), + ); + + expect(failure).toBeInstanceOf(ConnectionTransientError); + expect((yield* Ref.get(harness.storedTargets)).get(SSH_CONNECTION.environmentId)).toEqual( + SSH_RELAY_TARGET, + ); + expect( + (yield* SubscriptionRef.get(registry.entries)).get(SSH_CONNECTION.environmentId)?.target, + ).toEqual(SSH_RELAY_TARGET); + expect(yield* Ref.get(harness.releasedSessions)).toBe(0); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("moves durable streams to a replacement supervisor", () => Effect.gen(function* () { const replacement = new RelayConnectionTarget({ @@ -693,6 +869,64 @@ describe("EnvironmentRegistry", () => { }), ); + it.effect("removes a remembered relay route without disconnecting the selected SSH route", () => + Effect.gen(function* () { + const harness = yield* makeHarness( + [SSH_CONNECTION], + [SSH_PROFILE], + [[SSH_CONNECTION.connectionId, BEARER_CREDENTIAL]], + { initialRoutes: [SSH_RELAY_TARGET, SSH_CONNECTION] }, + ); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.removeRelayEnvironments(); + + expect((yield* Ref.get(harness.storedTargets)).get(SSH_CONNECTION.environmentId)).toEqual( + SSH_CONNECTION, + ); + expect([...(yield* Ref.get(harness.storedRoutes)).values()]).toEqual([SSH_CONNECTION]); + expect((yield* Ref.get(harness.storedProfiles)).has(SSH_CONNECTION.connectionId)).toBe( + true, + ); + expect((yield* Ref.get(harness.storedCredentials)).has(SSH_CONNECTION.connectionId)).toBe( + true, + ); + expect((yield* Ref.get(harness.storedRemoteTokens)).has(SSH_CONNECTION.environmentId)).toBe( + false, + ); + expect(yield* Ref.get(harness.cacheClears)).toEqual([]); + expect(yield* Ref.get(harness.ownedDataClears)).toEqual([]); + expect(yield* Ref.get(harness.disconnectedSshTargets)).toEqual([]); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + + it.effect("falls back to SSH when signing out while the relay route is selected", () => + Effect.gen(function* () { + const harness = yield* makeHarness( + [SSH_RELAY_TARGET], + [SSH_PROFILE], + [[SSH_CONNECTION.connectionId, BEARER_CREDENTIAL]], + { initialRoutes: [SSH_RELAY_TARGET, SSH_CONNECTION] }, + ); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + yield* registry.removeRelayEnvironments(); + + expect((yield* Ref.get(harness.storedTargets)).get(SSH_CONNECTION.environmentId)).toEqual( + SSH_CONNECTION, + ); + expect( + (yield* SubscriptionRef.get(registry.entries)).get(SSH_CONNECTION.environmentId)?.target, + ).toEqual(SSH_CONNECTION); + expect(yield* Ref.get(harness.cacheClears)).toEqual([]); + expect(yield* Ref.get(harness.ownedDataClears)).toEqual([]); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("keeps the runtime registered when durable removal fails", () => Effect.gen(function* () { const harness = yield* makeHarness([RELAY_TARGET], [], [], { @@ -776,6 +1010,12 @@ describe("EnvironmentRegistry", () => { (yield* SubscriptionRef.get(registry.entries)).get(TARGET.environmentId)?.target, ).toEqual(TARGET); + const routeError = yield* Effect.flip( + registry.selectRoute(TARGET.environmentId, connectionRouteId(TARGET)), + ); + expect(routeError._tag).toBe("PlatformEnvironmentRouteSelectionError"); + expect(routeError.message).toContain("cannot be selected"); + const error = yield* Effect.flip(registry.remove(TARGET.environmentId)); expect(error._tag).toBe("PlatformEnvironmentRemovalError"); expect( @@ -916,7 +1156,7 @@ describe("EnvironmentRegistry", () => { it.effect("removes all owned SSH state only on explicit removal", () => Effect.gen(function* () { const harness = yield* makeHarness( - [SSH_CONNECTION], + [SSH_RELAY_TARGET], [SSH_PROFILE], [ [ @@ -924,6 +1164,7 @@ describe("EnvironmentRegistry", () => { new BearerConnectionCredential({ token: "temporary-token" }), ], ], + { initialRoutes: [SSH_RELAY_TARGET, SSH_CONNECTION] }, ); yield* Effect.gen(function* () { diff --git a/packages/client-runtime/src/connection/registry.ts b/packages/client-runtime/src/connection/registry.ts index a3a36272f326..cca3a0ba79a8 100644 --- a/packages/client-runtime/src/connection/registry.ts +++ b/packages/client-runtime/src/connection/registry.ts @@ -28,8 +28,11 @@ import type { ConnectionAttemptError, ConnectionTarget, NetworkStatus, + PersistedConnectionTarget, + RelayConnectionTarget, SupervisorConnectionState, } from "./model.ts"; +import { connectionRouteId } from "./model.ts"; import * as Persistence from "../platform/persistence.ts"; import * as EnvironmentSupervisor from "./supervisor.ts"; import * as ConnectionDriver from "./driver.ts"; @@ -59,17 +62,55 @@ export class PlatformEnvironmentRemovalError extends Schema.TaggedErrorClass()( + "PlatformEnvironmentRouteSelectionError", + { + environmentId: EnvironmentId, + routeId: Schema.String, + }, +) { + override get message(): string { + return `Route ${this.routeId} cannot be selected for platform-managed environment ${this.environmentId}.`; + } +} + +export class ConnectionRouteNotRegisteredError extends Schema.TaggedErrorClass()( + "ConnectionRouteNotRegisteredError", + { + environmentId: EnvironmentId, + routeId: Schema.String, + }, +) { + override get message(): string { + return `Route ${this.routeId} is not registered for environment ${this.environmentId}.`; + } +} + export class EnvironmentRegistry extends Context.Service< EnvironmentRegistry, { readonly entries: SubscriptionRef.SubscriptionRef< ReadonlyMap >; + readonly routes: SubscriptionRef.SubscriptionRef< + ReadonlyMap> + >; readonly networkStatus: SubscriptionRef.SubscriptionRef; readonly start: Effect.Effect; readonly register: ( registration: ConnectionRegistration, ) => Effect.Effect; + readonly selectRoute: ( + environmentId: EnvironmentId, + routeId: string, + ) => Effect.Effect< + void, + | Persistence.ConnectionPersistenceError + | ConnectionAttemptError + | EnvironmentNotRegisteredError + | ConnectionRouteNotRegisteredError + | PlatformEnvironmentRouteSelectionError + >; readonly registerPlatform: (registration: PrimaryConnectionRegistration) => Effect.Effect; readonly reconcilePlatform: ( registrations: ReadonlyArray, @@ -137,24 +178,41 @@ export const make = Effect.gen(function* () { const wakeups = yield* ConnectionWakeups.ConnectionWakeups; const ssh = yield* ClientCapabilities.SshEnvironmentGateway; const persistedTargets = yield* storage.list; + const persistedRoutes = yield* storage.listRoutes; + const loadCatalogEntry = Effect.fn("EnvironmentRegistry.loadCatalogEntry")(function* ( + target: ConnectionTarget, + ) { + const profile = + target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget" + ? yield* profiles.get(target.connectionId) + : Option.none(); + return { target, profile } satisfies ConnectionCatalogEntry; + }); const initialEntries = new Map( yield* Effect.forEach( persistedTargets, - Effect.fn("EnvironmentRegistry.loadCatalogEntry")(function* (target) { - const profile = - target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget" - ? yield* profiles.get(target.connectionId) - : Option.none(); - return [ - target.environmentId, - { target, profile } satisfies ConnectionCatalogEntry, - ] as const; - }), + (target) => + loadCatalogEntry(target).pipe( + Effect.map((entry) => [target.environmentId, entry] as const), + ), { concurrency: "unbounded" }, ), ); + const initialRoutes = new Map>(); + for (const entry of yield* Effect.forEach(persistedRoutes, loadCatalogEntry, { + concurrency: "unbounded", + })) { + initialRoutes.set(entry.target.environmentId, [ + ...(initialRoutes.get(entry.target.environmentId) ?? []), + entry, + ]); + } const entries = yield* SubscriptionRef.make>(initialEntries); + const routes = + yield* SubscriptionRef.make>>( + initialRoutes, + ); const networkStatus = yield* SubscriptionRef.make(yield* connectivity.status); const serviceScopes = yield* SubscriptionRef.make< ReadonlyMap @@ -403,6 +461,57 @@ export const make = Effect.gen(function* () { next.set(environmentId, registration.target); return next; }); + yield* SubscriptionRef.update(routes, (current) => { + const next = new Map(current); + const existing = next.get(environmentId) ?? []; + next.set(environmentId, [ + ...existing.filter( + (candidate) => + connectionRouteId(candidate.target) !== connectionRouteId(entry.target), + ), + entry, + ]); + return next; + }); + yield* installEntryLocked(entry); + }), + ); + }); + + const selectRoute = Effect.fn("EnvironmentRegistry.selectRoute")(function* ( + environmentId: EnvironmentId, + routeId: string, + ) { + yield* withLeaseLock( + environmentId, + Effect.gen(function* () { + if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) { + return yield* new PlatformEnvironmentRouteSelectionError({ environmentId, routeId }); + } + yield* getEntry(environmentId); + const entry = (yield* SubscriptionRef.get(routes)) + .get(environmentId) + ?.find((candidate) => connectionRouteId(candidate.target) === routeId); + if (entry === undefined) { + return yield* new ConnectionRouteNotRegisteredError({ environmentId, routeId }); + } + if (entry.target._tag === "PrimaryConnectionTarget") { + return yield* new PlatformEnvironmentRouteSelectionError({ environmentId, routeId }); + } + const selected = (yield* SubscriptionRef.get(entries)).get(environmentId); + if (selected !== undefined && Equal.equals(selected, entry)) { + return; + } + + // Keep the current session active while credentials, transport, and the + // stable environment identity for the candidate route are verified. + yield* driver.prepare(entry); + yield* registrations.select(entry.target); + yield* Ref.update(persistedTargetsByEnvironment, (current) => { + const next = new Map(current); + next.set(environmentId, entry.target); + return next; + }); yield* installEntryLocked(entry); }), ); @@ -459,6 +568,11 @@ export const make = Effect.gen(function* () { ), ); } + yield* SubscriptionRef.update(routes, (current) => { + const next = new Map(current); + next.delete(target.environmentId); + return next; + }); yield* installEntryLocked(entry, { retainEquivalentRuntime: true }); }), @@ -551,10 +665,14 @@ export const make = Effect.gen(function* () { }); } const target = (yield* getEntry(environmentId)).target; - const profile = - target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget" - ? yield* profiles.get(target.connectionId) - : Option.none(); + const sshTargets = ((yield* SubscriptionRef.get(routes)).get(environmentId) ?? []).flatMap( + (entry) => + entry.target._tag === "SshConnectionTarget" && + Option.isSome(entry.profile) && + isSshConnectionProfile(entry.profile.value) + ? [entry.profile.value.target] + : [], + ); yield* registrations.remove(target); yield* Ref.update(persistedTargetsByEnvironment, (current) => { @@ -568,6 +686,11 @@ export const make = Effect.gen(function* () { next.delete(environmentId); return next; }); + yield* SubscriptionRef.update(routes, (current) => { + const next = new Map(current); + next.delete(environmentId); + return next; + }); yield* Effect.all( [ cache.clear(environmentId).pipe( @@ -583,37 +706,80 @@ export const make = Effect.gen(function* () { { concurrency: "unbounded", discard: true }, ); - if ( - target._tag === "SshConnectionTarget" && - Option.isSome(profile) && - isSshConnectionProfile(profile.value) - ) { - yield* ssh.disconnect(profile.value.target).pipe( - Effect.tapError((error) => - Effect.logWarning("Could not disconnect the managed SSH environment.", { - environmentId, - error, - }), + yield* Effect.forEach( + sshTargets, + (sshTarget) => + ssh.disconnect(sshTarget).pipe( + Effect.tapError((error) => + Effect.logWarning("Could not disconnect the managed SSH environment.", { + environmentId, + error, + }), + ), + Effect.ignore, ), - Effect.ignore, - ); - } + { discard: true }, + ); }), ); }); const removeRelayEnvironments = Effect.fn("EnvironmentRegistry.removeRelayEnvironments")( function* () { - const relayEnvironmentIds = [...(yield* SubscriptionRef.get(entries)).values()] - .filter((entry) => entry.target._tag === "RelayConnectionTarget") - .map((entry) => entry.target.environmentId); + const relayEnvironments = [...(yield* SubscriptionRef.get(routes))].flatMap( + ([environmentId, environmentRoutes]) => { + const relay = environmentRoutes.find( + (entry): entry is ConnectionCatalogEntry & { readonly target: RelayConnectionTarget } => + entry.target._tag === "RelayConnectionTarget", + ); + if (relay === undefined) return []; + const fallback = environmentRoutes.find( + ( + entry, + ): entry is ConnectionCatalogEntry & { + readonly target: Exclude; + } => + entry.target._tag === "BearerConnectionTarget" || + entry.target._tag === "SshConnectionTarget", + ); + return [{ environmentId, relay, fallback }]; + }, + ); yield* Effect.forEach( - relayEnvironmentIds, - (environmentId) => - remove(environmentId).pipe( - Effect.catchTag("EnvironmentNotRegisteredError", () => Effect.void), - ), + relayEnvironments, + ({ environmentId, relay, fallback }) => { + if (fallback === undefined) { + return remove(environmentId).pipe( + Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }), + ); + } + return withLeaseLock( + environmentId, + Effect.gen(function* () { + if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) { + return yield* new PlatformEnvironmentRemovalError({ environmentId }); + } + const selected = yield* getEntry(environmentId); + yield* registrations.removeRoute(relay.target, fallback.target); + yield* SubscriptionRef.update(routes, (current) => { + const next = new Map(current); + const remaining = (next.get(environmentId) ?? []).filter( + (entry) => connectionRouteId(entry.target) !== connectionRouteId(relay.target), + ); + next.set(environmentId, remaining); + return next; + }); + if (selected.target._tag !== "RelayConnectionTarget") return; + yield* Ref.update(persistedTargetsByEnvironment, (current) => { + const next = new Map(current); + next.set(environmentId, fallback.target); + return next; + }); + yield* installEntryLocked(fallback); + }), + ).pipe(Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void })); + }, { concurrency: "unbounded", discard: true, @@ -659,9 +825,11 @@ export const make = Effect.gen(function* () { return EnvironmentRegistry.of({ entries, + routes, networkStatus, start, register, + selectRoute, registerPlatform, reconcilePlatform, remove, diff --git a/packages/client-runtime/src/connection/resolver.test.ts b/packages/client-runtime/src/connection/resolver.test.ts index d71c0f9602bf..06f55c7caffa 100644 --- a/packages/client-runtime/src/connection/resolver.test.ts +++ b/packages/client-runtime/src/connection/resolver.test.ts @@ -23,6 +23,7 @@ import { import * as ConnectionCredentialStore from "./credentialStore.ts"; import { BearerConnectionTarget, + ConnectionBlockedError, ConnectionTransientError, PrimaryConnectionTarget, RelayConnectionTarget, @@ -445,6 +446,159 @@ describe("ConnectionResolver", () => { }), ); + it.effect("reuses a cached SSH bearer without requesting a new pairing credential", () => + Effect.gen(function* () { + const prepareBearerTokens = yield* Ref.make>([]); + const authorizeBearerTokens = yield* Ref.make>([]); + const target = new SshConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "SSH", + connectionId: "ssh-1", + }); + const profile = new SshConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + target: SSH_TARGET, + }); + const brokerLayer = yield* makeDependencies({ + profiles: [profile], + credentials: [ + [target.connectionId, new BearerConnectionCredential({ token: "cached-ssh" })], + ], + prepareSsh: (input) => + Ref.update(prepareBearerTokens, (values) => [...values, input.bearerToken]).pipe( + Effect.as({ + bootstrap: { + target: input.target, + httpBaseUrl: "http://127.0.0.1:4010", + wsBaseUrl: "ws://127.0.0.1:4010", + pairingToken: null, + }, + bearerToken: input.bearerToken ?? "unexpected-fresh-token", + }), + ), + authorizeBearer: (input) => + Ref.update(authorizeBearerTokens, (values) => [...values, input.bearerToken]).pipe( + Effect.as({ + environmentId: input.expectedEnvironmentId, + label: "SSH", + httpBaseUrl: input.httpBaseUrl, + socketUrl: "ws://127.0.0.1:4010/ws?wsTicket=cached", + httpAuthorization: { _tag: "Bearer" as const, token: input.bearerToken }, + }), + ), + }); + const broker = yield* ConnectionResolver.ConnectionResolver.pipe(Effect.provide(brokerLayer)); + + yield* broker.prepare(catalogEntry(target, Option.some(profile))); + + expect(yield* Ref.get(prepareBearerTokens)).toEqual(["cached-ssh"]); + expect(yield* Ref.get(authorizeBearerTokens)).toEqual(["cached-ssh"]); + }), + ); + + it.effect("replaces a cached SSH bearer only after authentication rejects it", () => + Effect.gen(function* () { + const prepareBearerTokens = yield* Ref.make>([]); + const authorizeBearerTokens = yield* Ref.make>([]); + const target = new SshConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "SSH", + connectionId: "ssh-1", + }); + const profile = new SshConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + target: SSH_TARGET, + }); + const brokerLayer = yield* makeDependencies({ + profiles: [profile], + credentials: [ + [target.connectionId, new BearerConnectionCredential({ token: "stale-ssh" })], + ], + prepareSsh: (input) => + Ref.update(prepareBearerTokens, (values) => [...values, input.bearerToken]).pipe( + Effect.as({ + bootstrap: { + target: input.target, + httpBaseUrl: "http://127.0.0.1:4010", + wsBaseUrl: "ws://127.0.0.1:4010", + pairingToken: input.bearerToken === undefined ? "pairing-token" : null, + }, + bearerToken: input.bearerToken ?? "fresh-ssh", + }), + ), + authorizeBearer: (input) => + Ref.update(authorizeBearerTokens, (values) => [...values, input.bearerToken]).pipe( + Effect.flatMap(() => + input.bearerToken === "stale-ssh" + ? Effect.fail( + new ConnectionBlockedError({ + reason: "authentication", + detail: "Credential rejected.", + }), + ) + : Effect.succeed({ + environmentId: input.expectedEnvironmentId, + label: "SSH", + httpBaseUrl: input.httpBaseUrl, + socketUrl: "ws://127.0.0.1:4010/ws?wsTicket=fresh", + httpAuthorization: { _tag: "Bearer" as const, token: input.bearerToken }, + }), + ), + ), + }); + const broker = yield* ConnectionResolver.ConnectionResolver.pipe(Effect.provide(brokerLayer)); + + yield* broker.prepare(catalogEntry(target, Option.some(profile))); + + expect(yield* Ref.get(prepareBearerTokens)).toEqual(["stale-ssh", undefined]); + expect(yield* Ref.get(authorizeBearerTokens)).toEqual(["stale-ssh", "fresh-ssh"]); + }), + ); + + it.effect("does not replace a cached SSH bearer after a transient failure", () => + Effect.gen(function* () { + const prepareBearerTokens = yield* Ref.make>([]); + const target = new SshConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "SSH", + connectionId: "ssh-1", + }); + const profile = new SshConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + target: SSH_TARGET, + }); + const brokerLayer = yield* makeDependencies({ + profiles: [profile], + credentials: [ + [target.connectionId, new BearerConnectionCredential({ token: "cached-ssh" })], + ], + prepareSsh: (input) => + Ref.update(prepareBearerTokens, (values) => [...values, input.bearerToken]).pipe( + Effect.flatMap(() => + Effect.fail( + new ConnectionTransientError({ + reason: "timeout", + detail: "SSH endpoint timed out.", + }), + ), + ), + ), + }); + const broker = yield* ConnectionResolver.ConnectionResolver.pipe(Effect.provide(brokerLayer)); + + const error = yield* Effect.flip(broker.prepare(catalogEntry(target, Option.some(profile)))); + + expect(error).toMatchObject({ _tag: "ConnectionTransientError", reason: "timeout" }); + expect(yield* Ref.get(prepareBearerTokens)).toEqual(["cached-ssh"]); + }), + ); + it.effect("classifies relay request timeouts as retryable connection failures", () => Effect.gen(function* () { const target = new RelayConnectionTarget({ diff --git a/packages/client-runtime/src/connection/resolver.ts b/packages/client-runtime/src/connection/resolver.ts index 3a5d5437a4e6..76db2da7f313 100644 --- a/packages/client-runtime/src/connection/resolver.ts +++ b/packages/client-runtime/src/connection/resolver.ts @@ -195,6 +195,7 @@ const makeRelayBroker = Effect.fn("clientRuntime.connection.broker.makeRelay")(f const makeSshBroker = Effect.fn("clientRuntime.connection.broker.makeSsh")(function* () { const profiles = yield* ConnectionProfileStore.ConnectionProfileStore; + const credentials = yield* ConnectionCredentialStore.ConnectionCredentialStore; const ssh = yield* ClientCapabilities.SshEnvironmentGateway; const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; @@ -218,33 +219,57 @@ const makeSshBroker = Effect.fn("clientRuntime.connection.broker.makeSsh")(funct actual: profile.environmentId, }); } - const prepared = yield* ssh.prepare({ - connectionId: target.connectionId, - expectedEnvironmentId: target.environmentId, - target: profile.target, - }); - yield* profiles.put( - new SshConnectionProfile({ - connectionId: profile.connectionId, - environmentId: profile.environmentId, - label: profile.label, - target: prepared.bootstrap.target, - }), + const prepareAndAuthorize = Effect.fn("clientRuntime.connection.broker.ssh.authorize")( + function* (bearerToken?: string) { + const prepared = yield* ssh.prepare({ + connectionId: target.connectionId, + expectedEnvironmentId: target.environmentId, + target: profile.target, + ...(bearerToken === undefined ? {} : { bearerToken }), + }); + yield* profiles.put( + new SshConnectionProfile({ + connectionId: profile.connectionId, + environmentId: profile.environmentId, + label: profile.label, + target: prepared.bootstrap.target, + }), + ); + const authorized = yield* remote.authorizeBearer({ + expectedEnvironmentId: target.environmentId, + httpBaseUrl: prepared.bootstrap.httpBaseUrl, + wsBaseUrl: prepared.bootstrap.wsBaseUrl, + bearerToken: prepared.bearerToken, + }); + if (bearerToken === undefined) { + yield* credentials.put( + target.connectionId, + new BearerConnectionCredential({ token: prepared.bearerToken }), + ); + } + return { + environmentId: authorized.environmentId, + label: authorized.label, + httpBaseUrl: authorized.httpBaseUrl, + socketUrl: authorized.socketUrl, + httpAuthorization: authorized.httpAuthorization, + target, + } satisfies PreparedConnection; + }, ); - const authorized = yield* remote.authorizeBearer({ - expectedEnvironmentId: target.environmentId, - httpBaseUrl: prepared.bootstrap.httpBaseUrl, - wsBaseUrl: prepared.bootstrap.wsBaseUrl, - bearerToken: prepared.bearerToken, - }); - return { - environmentId: authorized.environmentId, - label: authorized.label, - httpBaseUrl: authorized.httpBaseUrl, - socketUrl: authorized.socketUrl, - httpAuthorization: authorized.httpAuthorization, - target, - } satisfies PreparedConnection; + + const credential = yield* credentials.get(target.connectionId); + if (Option.isSome(credential) && isBearerCredential(credential.value)) { + return yield* prepareAndAuthorize(credential.value.token).pipe( + Effect.catchTags({ + ConnectionBlockedError: (error) => + error.reason === "authentication" + ? credentials.remove(target.connectionId).pipe(Effect.andThen(prepareAndAuthorize())) + : Effect.fail(error), + }), + ); + } + return yield* prepareAndAuthorize(); }); }); diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 5e50c44d9610..e8a13a9c8a32 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -20,6 +20,7 @@ import { ConnectionTransientError, PrimaryConnectionTarget, RelayConnectionTarget, + SshConnectionTarget, type ConnectionAttemptError, type ConnectionTarget, type NetworkStatus, @@ -52,6 +53,15 @@ const RELAY_ENTRY: ConnectionCatalogEntry = { profile: Option.none(), }; +const SSH_ENTRY: ConnectionCatalogEntry = { + target: new SshConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + connectionId: "ssh:test-environment", + }), + profile: Option.none(), +}; + const PREPARED_CONNECTION: PreparedConnection = { environmentId: TARGET.environmentId, label: TARGET.label, @@ -188,7 +198,10 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: ), Layer.succeed( ConnectionDriver.ConnectionDriver, - ConnectionDriver.ConnectionDriver.of({ connect }), + ConnectionDriver.ConnectionDriver.of({ + prepare: (entry) => prepare(entry.target), + connect, + }), ), ); @@ -465,6 +478,31 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("allows SSH launch longer than the standard setup timeout", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + prepare: () => Effect.never, + }); + const supervisor = yield* EnvironmentSupervisor.make(SSH_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connecting" && state.stage === "preparing", + ); + yield* TestClock.adjust("119 seconds"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connecting"); + + yield* TestClock.adjust("1 second"); + const retrying = yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + expect(retrying.lastFailure?.reason).toBe("timeout"); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("converts unexpected driver defects into retryable failures", () => 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 85fda10ef1a7..4c1c92adc915 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -31,6 +31,7 @@ import * as ConnectionWakeups from "./wakeups.ts"; const RETRY_DELAYS_MS = [3_000, 4_000, 8_000, 16_000] as const; const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; +const SSH_CONNECTION_ESTABLISHMENT_TIMEOUT = "120 seconds"; const CONNECTION_PROBE_TIMEOUT = "15 seconds"; const MOBILE_CONNECTION_PROBE_TIMEOUT = "3 seconds"; const BACKOFF_RESET_AFTER_MS = 30_000; @@ -513,9 +514,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( }), ), ), - Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe( - Effect.as({ _tag: "TimedOut" }), - ), + Effect.sleep( + target._tag === "SshConnectionTarget" + ? SSH_CONNECTION_ESTABLISHMENT_TIMEOUT + : CONNECTION_ESTABLISHMENT_TIMEOUT, + ).pipe(Effect.as({ _tag: "TimedOut" })), ]); if (establishment._tag === "Interrupted") { diff --git a/packages/client-runtime/src/platform/capabilities.ts b/packages/client-runtime/src/platform/capabilities.ts index a20b7d404b2f..c1339d1c7e93 100644 --- a/packages/client-runtime/src/platform/capabilities.ts +++ b/packages/client-runtime/src/platform/capabilities.ts @@ -60,6 +60,7 @@ export class SshEnvironmentGateway extends Context.Service< readonly connectionId: string; readonly expectedEnvironmentId: EnvironmentId; readonly target: DesktopSshEnvironmentTarget; + readonly bearerToken?: string; }) => Effect.Effect; readonly disconnect: ( target: DesktopSshEnvironmentTarget, diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index 14a322bcd303..73d4b5d81e3b 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -12,14 +12,17 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import type { ConnectionRegistration } from "../connection/catalog.ts"; -import type { ConnectionTarget } from "../connection/model.ts"; +import type { ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts"; export class ConnectionPersistenceError extends Schema.TaggedErrorClass()( "ConnectionPersistenceError", { operation: Schema.Literals([ "list-targets", + "list-routes", "register-connection", + "select-connection-route", + "remove-connection-route", "remove-connection", "load-shell", "save-shell", @@ -42,6 +45,7 @@ export class ConnectionTargetStore extends Context.Service< ConnectionTargetStore, { readonly list: Effect.Effect, ConnectionPersistenceError>; + readonly listRoutes: Effect.Effect, ConnectionPersistenceError>; } >()("@t3tools/client-runtime/platform/persistence/ConnectionTargetStore") {} @@ -51,6 +55,13 @@ export class ConnectionRegistrationStore extends Context.Service< readonly register: ( registration: ConnectionRegistration, ) => Effect.Effect; + readonly select: ( + target: PersistedConnectionTarget, + ) => Effect.Effect; + readonly removeRoute: ( + target: PersistedConnectionTarget, + fallback: PersistedConnectionTarget, + ) => Effect.Effect; readonly remove: (target: ConnectionTarget) => Effect.Effect; } >()("@t3tools/client-runtime/platform/persistence/ConnectionRegistrationStore") {} diff --git a/packages/client-runtime/src/platform/storageDocument.test.ts b/packages/client-runtime/src/platform/storageDocument.test.ts index 8ad6b81e12b4..89459aa8a6a9 100644 --- a/packages/client-runtime/src/platform/storageDocument.test.ts +++ b/packages/client-runtime/src/platform/storageDocument.test.ts @@ -1,5 +1,6 @@ import { EnvironmentId } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Schema from "effect/Schema"; import * as TokenStore from "../authorization/tokenStore.ts"; import { @@ -17,7 +18,10 @@ import { } from "../connection/model.ts"; import { EMPTY_CONNECTION_CATALOG_DOCUMENT, + ConnectionCatalogDocument, + connectionRoutes, registerConnectionInCatalog, + removeConnectionRouteFromCatalog, removeConnectionFromCatalog, } from "./storageDocument.ts"; @@ -52,6 +56,19 @@ const REMOTE_TOKEN = new TokenStore.RemoteDpopAccessToken({ }); describe("ConnectionCatalogDocument", () => { + it("treats targets from a legacy catalog as its initial routes", () => { + const decoded = Schema.decodeUnknownSync(ConnectionCatalogDocument)({ + schemaVersion: 1, + targets: [BEARER_TARGET], + profiles: [BEARER_PROFILE], + credentials: [], + remoteDpopTokens: [], + }); + + expect(decoded.routes).toEqual([]); + expect(connectionRoutes(decoded)).toEqual([BEARER_TARGET]); + }); + it("registers a bearer connection as one catalog mutation", () => { const document = registerConnectionInCatalog( EMPTY_CONNECTION_CATALOG_DOCUMENT, @@ -63,6 +80,7 @@ describe("ConnectionCatalogDocument", () => { ); expect(document.targets).toEqual([BEARER_TARGET]); + expect(document.routes).toEqual([BEARER_TARGET]); expect(document.profiles).toEqual([BEARER_PROFILE]); expect(document.credentials).toEqual([ { @@ -72,7 +90,7 @@ describe("ConnectionCatalogDocument", () => { ]); }); - it("replaces obsolete connection metadata without discarding a reusable DPoP token", () => { + it("selects a newly registered route without discarding other routes", () => { const bearer = registerConnectionInCatalog( { ...EMPTY_CONNECTION_CATALOG_DOCUMENT, @@ -94,8 +112,14 @@ describe("ConnectionCatalogDocument", () => { ); expect(relay.targets).toEqual([relayTarget]); - expect(relay.profiles).toEqual([]); - expect(relay.credentials).toEqual([]); + expect(relay.routes).toEqual([BEARER_TARGET, relayTarget]); + expect(relay.profiles).toEqual([BEARER_PROFILE]); + expect(relay.credentials).toEqual([ + { + connectionId: BEARER_TARGET.connectionId, + credential: BEARER_CREDENTIAL, + }, + ]); expect(relay.remoteDpopTokens).toEqual([REMOTE_TOKEN]); }); @@ -117,6 +141,34 @@ describe("ConnectionCatalogDocument", () => { ); }); + it("removes a relay route without discarding another route for the environment", () => { + const relayTarget = new RelayConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "Remote", + }); + const registered = registerConnectionInCatalog( + registerConnectionInCatalog( + { + ...EMPTY_CONNECTION_CATALOG_DOCUMENT, + remoteDpopTokens: [REMOTE_TOKEN], + }, + new BearerConnectionRegistration({ + target: BEARER_TARGET, + profile: BEARER_PROFILE, + credential: BEARER_CREDENTIAL, + }), + ), + new RelayConnectionRegistration({ target: relayTarget }), + ); + + expect(removeConnectionRouteFromCatalog(registered, relayTarget, BEARER_TARGET)).toEqual({ + ...registered, + targets: [BEARER_TARGET], + routes: [BEARER_TARGET], + remoteDpopTokens: [], + }); + }); + it("persists the normalized SSH profile beside its target", () => { const target = new SshConnectionTarget({ environmentId: ENVIRONMENT_ID, @@ -140,7 +192,65 @@ describe("ConnectionCatalogDocument", () => { ); expect(document.targets).toEqual([target]); + expect(document.routes).toEqual([target]); expect(document.profiles).toEqual([profile]); expect(document.credentials).toEqual([]); }); + + it("persists an SSH bearer credential when provisioning supplies one", () => { + const target = new SshConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "SSH", + connectionId: "ssh-1", + }); + const profile = new SshConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + target: { + alias: "devbox", + hostname: "devbox.example.test", + username: "developer", + port: 22, + }, + }); + const credential = new BearerConnectionCredential({ token: "ssh-bearer" }); + const document = registerConnectionInCatalog( + EMPTY_CONNECTION_CATALOG_DOCUMENT, + new SshConnectionRegistration({ target, profile, credential }), + ); + + expect(document.credentials).toEqual([{ connectionId: target.connectionId, credential }]); + }); + + it("preserves an SSH bearer credential when re-registration supplies no replacement", () => { + const target = new SshConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: "SSH", + connectionId: "ssh-1", + }); + const profile = new SshConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + target: { + alias: "devbox", + hostname: "devbox.example.test", + username: "developer", + port: 22, + }, + }); + const credential = new BearerConnectionCredential({ token: "ssh-bearer" }); + const registered = registerConnectionInCatalog( + EMPTY_CONNECTION_CATALOG_DOCUMENT, + new SshConnectionRegistration({ target, profile, credential }), + ); + + const reRegistered = registerConnectionInCatalog( + registered, + new SshConnectionRegistration({ target, profile }), + ); + + expect(reRegistered.credentials).toEqual([{ connectionId: target.connectionId, credential }]); + }); }); diff --git a/packages/client-runtime/src/platform/storageDocument.ts b/packages/client-runtime/src/platform/storageDocument.ts index 0ba55dfa2fbc..b2db59de2cdb 100644 --- a/packages/client-runtime/src/platform/storageDocument.ts +++ b/packages/client-runtime/src/platform/storageDocument.ts @@ -1,3 +1,4 @@ +import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { @@ -5,7 +6,11 @@ import { ConnectionCredential, ConnectionProfile, } from "../connection/catalog.ts"; -import { type ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts"; +import { + type ConnectionTarget, + PersistedConnectionTarget, + connectionRouteId, +} from "../connection/model.ts"; import * as TokenStore from "../authorization/tokenStore.ts"; export const StoredConnectionCredential = Schema.Struct({ @@ -17,6 +22,9 @@ export type StoredConnectionCredential = typeof StoredConnectionCredential.Type; export const ConnectionCatalogDocument = Schema.Struct({ schemaVersion: Schema.Literal(1), targets: Schema.Array(PersistedConnectionTarget), + routes: Schema.Array(PersistedConnectionTarget).pipe( + Schema.withDecodingDefaultKey(Effect.succeed([])), + ), profiles: Schema.Array(ConnectionProfile), credentials: Schema.Array(StoredConnectionCredential), remoteDpopTokens: Schema.Array(TokenStore.RemoteDpopAccessToken), @@ -26,6 +34,7 @@ export type ConnectionCatalogDocument = typeof ConnectionCatalogDocument.Type; export const EMPTY_CONNECTION_CATALOG_DOCUMENT: ConnectionCatalogDocument = Object.freeze({ schemaVersion: 1, targets: [], + routes: [], profiles: [], credentials: [], remoteDpopTokens: [], @@ -59,19 +68,21 @@ function connectionIdOf(target: ConnectionTarget): string | null { } } -function removeConnectionMetadata( +export function connectionRoutes( + document: ConnectionCatalogDocument, +): ReadonlyArray { + return document.routes.length === 0 && document.targets.length > 0 + ? document.targets + : document.routes; +} + +function removeRouteMetadata( document: ConnectionCatalogDocument, target: ConnectionTarget, - removeRemoteToken: boolean, ): ConnectionCatalogDocument { const connectionId = connectionIdOf(target); return { ...document, - targets: removeCatalogValue( - document.targets, - (value) => value.environmentId, - target.environmentId, - ), profiles: connectionId === null ? document.profiles @@ -80,13 +91,6 @@ function removeConnectionMetadata( connectionId === null ? document.credentials : removeCatalogValue(document.credentials, (value) => value.connectionId, connectionId), - remoteDpopTokens: removeRemoteToken - ? removeCatalogValue( - document.remoteDpopTokens, - (value) => value.environmentId, - target.environmentId, - ) - : document.remoteDpopTokens, }; } @@ -95,14 +99,21 @@ export function registerConnectionInCatalog( registration: ConnectionRegistration, ): ConnectionCatalogDocument { const target = registration.target; - const previous = document.targets.find( - (candidate) => candidate.environmentId === target.environmentId, + const routes = connectionRoutes(document); + const previous = routes.find( + (candidate) => connectionRouteId(candidate) === connectionRouteId(target), ); - const cleaned = - previous === undefined ? document : removeConnectionMetadata(document, previous, false); + const previousCredential = + registration._tag === "SshConnectionRegistration" && registration.credential === undefined + ? document.credentials.find( + (value) => value.connectionId === registration.target.connectionId, + ) + : undefined; + const cleaned = previous === undefined ? document : removeRouteMetadata(document, previous); const next: ConnectionCatalogDocument = { ...cleaned, targets: replaceCatalogValue(cleaned.targets, (value) => value.environmentId, target), + routes: replaceCatalogValue(routes, connectionRouteId, target), }; switch (registration._tag) { @@ -129,6 +140,19 @@ export function registerConnectionInCatalog( (value) => value.connectionId, registration.profile, ), + credentials: + registration.credential === undefined + ? previousCredential === undefined + ? next.credentials + : replaceCatalogValue( + next.credentials, + (value) => value.connectionId, + previousCredential, + ) + : replaceCatalogValue(next.credentials, (value) => value.connectionId, { + connectionId: registration.target.connectionId, + credential: registration.credential, + }), }; } } @@ -137,5 +161,74 @@ export function removeConnectionFromCatalog( document: ConnectionCatalogDocument, target: ConnectionTarget, ): ConnectionCatalogDocument { - return removeConnectionMetadata(document, target, true); + const environmentRoutes = connectionRoutes(document).filter( + (candidate) => candidate.environmentId === target.environmentId, + ); + const connectionIds = new Set( + environmentRoutes + .map(connectionIdOf) + .filter((connectionId): connectionId is string => connectionId !== null), + ); + return { + ...document, + targets: removeCatalogValue( + document.targets, + (value) => value.environmentId, + target.environmentId, + ), + routes: connectionRoutes(document).filter( + (candidate) => candidate.environmentId !== target.environmentId, + ), + profiles: document.profiles.filter((profile) => !connectionIds.has(profile.connectionId)), + credentials: document.credentials.filter( + (credential) => !connectionIds.has(credential.connectionId), + ), + remoteDpopTokens: removeCatalogValue( + document.remoteDpopTokens, + (value) => value.environmentId, + target.environmentId, + ), + }; +} + +export function removeConnectionRouteFromCatalog( + document: ConnectionCatalogDocument, + target: ConnectionCatalogDocument["routes"][number], + fallback: ConnectionCatalogDocument["routes"][number], +): ConnectionCatalogDocument { + const cleaned = removeRouteMetadata(document, target); + const selected = document.targets.find( + (candidate) => candidate.environmentId === target.environmentId, + ); + return { + ...cleaned, + targets: + selected !== undefined && connectionRouteId(selected) === connectionRouteId(target) + ? replaceCatalogValue(cleaned.targets, (value) => value.environmentId, fallback) + : cleaned.targets, + routes: connectionRoutes(cleaned).filter( + (candidate) => + candidate.environmentId !== target.environmentId || + connectionRouteId(candidate) !== connectionRouteId(target), + ), + remoteDpopTokens: + target._tag === "RelayConnectionTarget" + ? removeCatalogValue( + cleaned.remoteDpopTokens, + (value) => value.environmentId, + target.environmentId, + ) + : cleaned.remoteDpopTokens, + }; +} + +export function selectConnectionRouteInCatalog( + document: ConnectionCatalogDocument, + target: ConnectionCatalogDocument["routes"][number], +): ConnectionCatalogDocument { + return { + ...document, + targets: replaceCatalogValue(document.targets, (value) => value.environmentId, target), + routes: connectionRoutes(document), + }; } diff --git a/packages/client-runtime/src/state/connections.ts b/packages/client-runtime/src/state/connections.ts index 6dfa5001a486..c894aa51802d 100644 --- a/packages/client-runtime/src/state/connections.ts +++ b/packages/client-runtime/src/state/connections.ts @@ -50,6 +50,27 @@ export function createEnvironmentCatalogAtoms( Option.getOrElse(AsyncResult.value(get(catalogAtom)), () => EMPTY_ENVIRONMENT_CATALOG_STATE), ).pipe(Atom.withLabel("environment-catalog-value")); + const routesAtom = runtime.atom( + Stream.unwrap( + EnvironmentRegistry.EnvironmentRegistry.pipe( + Effect.map((registry) => SubscriptionRef.changes(registry.routes)), + ), + ), + { + initialValue: new Map< + EnvironmentIdType, + ReadonlyArray + >() as ReadonlyMap>, + }, + ); + + const routesValueAtom = Atom.make((get) => + Option.getOrElse( + AsyncResult.value(get(routesAtom)), + () => new Map>(), + ), + ).pipe(Atom.withLabel("environment-catalog-routes-value")); + const networkStatusAtom = runtime.atom( Stream.unwrap( EnvironmentRegistry.EnvironmentRegistry.pipe( @@ -97,6 +118,15 @@ export function createEnvironmentCatalogAtoms( Effect.flatMap((registry) => registry.remove(environmentId)), ), }); + const selectRoute = createRuntimeCommand(runtime, { + label: "environment-catalog:select-route", + scheduler: commandScheduler, + concurrency: serial, + execute: (input: { readonly environmentId: EnvironmentIdType; readonly routeId: string }) => + EnvironmentRegistry.EnvironmentRegistry.pipe( + Effect.flatMap((registry) => registry.selectRoute(input.environmentId, input.routeId)), + ), + }); const removeRelayEnvironments = createRuntimeCommand(runtime, { label: "environment-catalog:remove-relay-environments", scheduler: commandScheduler, @@ -119,10 +149,13 @@ export function createEnvironmentCatalogAtoms( return { catalogAtom, catalogValueAtom, + routesAtom, + routesValueAtom, networkStatusAtom, networkStatusValueAtom, stateAtom, register, + selectRoute, remove, removeRelayEnvironments, retryNow, diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index 392885b640c1..46bfd0dcda90 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -188,7 +188,15 @@ describe("ssh tunnel scripts", () => { assert.include(buildRemoteLaunchScript(), 'kill "$REMOTE_PID" 2>/dev/null || true'); assert.include(buildRemoteLaunchScript(), "wait_ready"); assert.include(buildRemoteLaunchScript(), '"$RUNNER_FILE" serve --host 127.0.0.1'); - assert.include(buildRemoteLaunchScript(), '--base-dir "$DEFAULT_SERVER_HOME"'); + assert.notInclude(buildRemoteLaunchScript(), '--base-dir "$DEFAULT_SERVER_HOME"'); + assert.include(buildRemoteLaunchScript(), "discover_running_runtime()"); + assert.include(buildRemoteLaunchScript(), "systemctl --user show t3code.service"); + assert.notInclude(buildRemoteLaunchScript(), 'fs.readdirSync("/proc")'); + assert.include(buildRemoteLaunchScript(), "knownBaseDirPath"); + assert.include(buildRemoteLaunchScript(), '"server-runtime.json"'); + assert.include(buildRemoteLaunchScript(), 'origin.protocol !== "http:"'); + assert.notInclude(buildRemoteLaunchScript(), "AbortSignal.timeout(2500)"); + assert.include(buildRemoteLaunchScript(), "Existing remote T3 server did not become ready"); assert.notInclude(buildRemoteLaunchScript(), "server-home"); assert.include(buildRemoteLaunchScript(), "Remote T3 server did not become ready"); assert.include(buildRemoteLaunchScript(), 'wait_ready "60000"'); @@ -197,17 +205,44 @@ describe("ssh tunnel scripts", () => { assert.include(buildRemoteLaunchScript({ packageSpec: "t3@nightly" }), "t3@nightly"); assert.include( buildRemotePairingScript(target), - '"$RUNNER_FILE" auth pairing create --base-dir "$PAIRING_BASE_DIR" --json', + '"$RUNNER_FILE" pair --base-dir "$PAIRING_BASE_DIR" >"$PAIR_OUTPUT_FILE" 2>&1', ); - assert.include(buildRemotePairingScript(target), 'PAIRING_BASE_DIR="$DEFAULT_SERVER_HOME"'); + assert.include(buildRemotePairingScript(target), 'if [ -z "$PAIRING_BASE_DIR" ]'); + assert.notInclude(buildRemotePairingScript(target), '"$RUNNER_FILE" pair >'); + assert.include(buildRemotePairingScript(target), "grep -q 'database is locked'"); + assert.include(buildRemotePairingScript(target), 'while [ "$PAIR_ATTEMPT" -lt 5 ]'); + assert.include(buildRemotePairingScript(target), 'RUNNER_NEXT="$STATE_DIR/run-t3.next.$$"'); + assert.include(buildRemotePairingScript(target), 'cat >"$RUNNER_NEXT"'); + assert.include(buildRemotePairingScript(target), 'mv -f "$RUNNER_NEXT" "$RUNNER_FILE"'); + assert.notInclude(buildRemotePairingScript(target), 'cat >"$RUNNER_FILE"'); + for (const script of [ + buildRemoteLaunchScript(), + buildRemotePairingScript(target), + buildRemoteStopScript(target), + ]) { + assert.include(script, "acquire_state_lock()"); + assert.include(script, "release_state_lock()"); + assert.include(script, "flock -w 45 9"); + assert.include(script, 'ln "$STATE_LOCK_OWNER_FILE" "$STATE_LOCK_LINK"'); + assert.include(script, 'kill -0 "$STATE_LOCK_OWNER_PID"'); + assert.include(script, 'STATE_LOCK_WAIT_COUNT" -ge 450'); + assert.include(script, "acquire_state_lock"); + } + assert.include(buildRemotePairingScript(target), 'line.startsWith("Token: ")'); + assert.include(buildRemotePairingScript(target), 'BASE_DIR_FILE="$STATE_DIR/base-dir"'); assert.notInclude(buildRemotePairingScript(target), "server-home"); assert.include(buildRemotePairingScript(target, { packageSpec: "t3@nightly" }), "t3@nightly"); - assert.include( - buildRemoteStopScript(target), - 'if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ]', - ); + assert.include(buildRemoteStopScript(target), 'if [ "$REMOTE_MANAGED" = "external" ]'); assert.include(buildRemoteStopScript(target), 'kill "$REMOTE_PID" 2>/dev/null || true'); assert.include(buildRemoteStopScript(target), 'rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE"'); + assert.include(buildRemoteLaunchScript(), "< /dev/null 9>&- &"); + assert.include(buildRemoteLaunchScript(), 'mv "$DISCOVERED_BASE_DIR_FILE" "$BASE_DIR_FILE"'); + assert.include(buildRemoteLaunchScript(), 'REMOTE_PORT="$PREVIOUS_REMOTE_PORT"'); + assert.include(buildRemoteLaunchScript(), 'if [ -n "$LAUNCHED_PID" ]; then'); + assert.notInclude( + buildRemoteLaunchScript(), + 'discover_running_runtime() {\n rm -f "$BASE_DIR_FILE"', + ); assert.include( buildRemoteLaunchScript(), 'DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json"', @@ -236,6 +271,59 @@ describe("ssh tunnel scripts", () => { ); }); + it("embeds syntactically valid Node scripts", () => { + const target = { + alias: "devbox", + hostname: "devbox.example.com", + username: "julius", + port: 2222, + } as const; + + for (const script of [buildRemoteLaunchScript(), buildRemotePairingScript(target)]) { + const embeddedScripts = [...script.matchAll(/<<'NODE'\n([\s\S]*?)\nNODE/gu)]; + assert.isAbove(embeddedScripts.length, 0); + for (const match of embeddedScripts) { + assert.doesNotThrow(() => Function(match[1]!)); + } + } + }); + + it.effect("emits syntactically valid remote shell scripts", () => { + const target = { + alias: "devbox", + hostname: "devbox.example.com", + username: "julius", + port: 2222, + } as const; + + return Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + for (const script of [ + buildRemoteLaunchScript(), + buildRemotePairingScript(target), + buildRemoteStopScript(target), + ]) { + assert.notMatch(script, /@@[A-Z0-9_]+@@/u); + const handle = yield* spawner.spawn(ChildProcess.make("sh", ["-n"])); + yield* Stream.make(new TextEncoder().encode(script)).pipe(Stream.run(handle.stdin)); + const [stderr, exitCode] = yield* Effect.all( + [ + handle.stderr.pipe( + Stream.decodeText(), + Stream.runFold( + () => "", + (accumulator, chunk) => accumulator + chunk, + ), + ), + handle.exitCode, + ], + { concurrency: "unbounded" }, + ); + assert.equal(exitCode, 0, stderr); + } + }).pipe(Effect.provide(NodeServices.layer)); + }); + it.effect("accepts launch JSON after remote shell startup noise", () => { const target = { alias: "devbox", @@ -446,4 +534,82 @@ describe("ssh tunnel scripts", () => { assert.equal(tunnelKillCount, 1); }).pipe(Effect.provide(layer), Effect.scoped); }); + + it.effect("serializes pairing commands for concurrent ensures of one SSH target", () => { + let activePairingCommands = 0; + let maximumActivePairingCommands = 0; + let pairingCommandCount = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.sync(() => { + const args = commandArgs(command); + if (args.includes("-N")) { + return makeRunningProcess(() => {}); + } + if (args.includes("sh") && args.includes("--")) { + return makeSuccessfulProcess('{"remotePort":3773,"serverKind":"external"}\n'); + } + if (args.includes("sh")) { + pairingCommandCount += 1; + const stdout = Stream.make( + new TextEncoder().encode(`{"credential":"PAIR-${pairingCommandCount}"}\n`), + ); + return ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(200 + pairingCommandCount), + stdout, + stderr: Stream.empty, + all: stdout, + exitCode: Effect.gen(function* () { + activePairingCommands += 1; + maximumActivePairingCommands = Math.max( + maximumActivePairingCommands, + activePairingCommands, + ); + yield* Effect.yieldNow; + activePairingCommands -= 1; + return ChildProcessSpawner.ExitCode(0); + }), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + stdin: Sink.drain, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }); + } + return makeSuccessfulProcess("\n"); + }), + ); + const layer = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(HttpClient.HttpClient, testHttpClient), + Layer.succeed(NetService.NetService, testNetService), + SshPasswordPrompt.disabledLayer, + SshEnvironmentManager.layer(), + ); + const target = { + alias: "devbox", + hostname: "devbox.example.com", + username: "julius", + port: 2222, + } as const; + + return Effect.gen(function* () { + const manager = yield* SshEnvironmentManager; + const results = yield* Effect.all( + [ + manager.ensureEnvironment(target, { issuePairingToken: true }), + manager.ensureEnvironment(target, { issuePairingToken: true }), + ], + { concurrency: "unbounded" }, + ); + + assert.equal(pairingCommandCount, 2); + assert.equal(maximumActivePairingCommands, 1); + assert.deepEqual(results.map((result) => result.pairingToken).toSorted(), [ + "PAIR-1", + "PAIR-2", + ]); + }).pipe(Effect.provide(layer), Effect.scoped); + }); }); diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index 0afd1007f8ed..7b0798e23c5e 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -17,6 +17,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; @@ -56,7 +57,7 @@ const SSH_READY_PROBE_TIMEOUT_MS = 1_000; const TUNNEL_SHUTDOWN_TIMEOUT_MS = 2_000; const REMOTE_READY_TIMEOUT_MS = 60_000; const REMOTE_LAUNCH_TIMEOUT_MS = 90_000; -const REMOTE_REUSE_READY_TIMEOUT_MS = 2_000; +const REMOTE_REUSE_READY_TIMEOUT_MS = 20_000; export interface RemoteT3RunnerOptions { readonly packageSpec?: string; @@ -452,6 +453,57 @@ printf 'Remote host is missing the t3 CLI and could not install @@T3_PACKAGE_SPE exit 1 `; +const REMOTE_STATE_LOCK_SCRIPT = `acquire_state_lock() { + STATE_LOCK_FILE="$STATE_DIR/operation.lock" + STATE_LOCK_LINK="$STATE_DIR/operation.lock.link" + STATE_LOCK_OWNER_FILE="$STATE_DIR/operation.lock.owner.$$" + STATE_LOCK_MODE="" + if command -v flock >/dev/null 2>&1; then + exec 9>"$STATE_LOCK_FILE" + if ! flock -w 45 9; then + printf 'Timed out waiting for another T3 SSH operation to finish.\n' >&2 + return 1 + fi + STATE_LOCK_MODE="flock" + return 0 + fi + + STATE_LOCK_WAIT_COUNT=0 + printf '%s\n' "$$" >"$STATE_LOCK_OWNER_FILE" + while ! ln "$STATE_LOCK_OWNER_FILE" "$STATE_LOCK_LINK" 2>/dev/null; do + STATE_LOCK_OWNER_PID="$(cat "$STATE_LOCK_LINK" 2>/dev/null || true)" + if [ -n "$STATE_LOCK_OWNER_PID" ]; then + if ! kill -0 "$STATE_LOCK_OWNER_PID" 2>/dev/null; then + rm -f "$STATE_LOCK_LINK" + continue + fi + fi + STATE_LOCK_WAIT_COUNT=$((STATE_LOCK_WAIT_COUNT + 1)) + if [ "$STATE_LOCK_WAIT_COUNT" -ge 450 ]; then + printf 'Timed out waiting for another T3 SSH operation to finish.\n' >&2 + rm -f "$STATE_LOCK_OWNER_FILE" + return 1 + fi + sleep 0.1 + done + rm -f "$STATE_LOCK_OWNER_FILE" + STATE_LOCK_MODE="link" +} + +release_state_lock() { + if [ "\${STATE_LOCK_MODE:-}" = "flock" ]; then + flock -u 9 2>/dev/null || true + exec 9>&- + elif [ "\${STATE_LOCK_MODE:-}" = "link" ]; then + STATE_LOCK_OWNER_PID="$(cat "$STATE_LOCK_LINK" 2>/dev/null || true)" + if [ "$STATE_LOCK_OWNER_PID" = "$$" ]; then + rm -f "$STATE_LOCK_LINK" + fi + fi + rm -f "\${STATE_LOCK_OWNER_FILE:-}" + STATE_LOCK_MODE="" +}`; + export const REMOTE_LAUNCH_SCRIPT = `set -eu @@T3_NODE_ENV_SCRIPT@@ STATE_KEY="$1" @@ -461,14 +513,36 @@ DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json" PORT_FILE="$STATE_DIR/port" PID_FILE="$STATE_DIR/pid" MANAGED_FILE="$STATE_DIR/managed" +BASE_DIR_FILE="$STATE_DIR/base-dir" LOG_FILE="$STATE_DIR/server.log" RUNNER_FILE="$STATE_DIR/run-t3.sh" RUNNER_NEXT="$STATE_DIR/run-t3.next.$$" +DISCOVERED_BASE_DIR_FILE="$STATE_DIR/base-dir.next.$$" mkdir -p "$STATE_DIR" +umask 077 +@@T3_STATE_LOCK_SCRIPT@@ +LAUNCHED_PID="" cleanup_runner_next() { - rm -f "$RUNNER_NEXT" + rm -f "$RUNNER_NEXT" "$DISCOVERED_BASE_DIR_FILE" + release_state_lock +} +abort_remote_launch() { + if [ -n "$LAUNCHED_PID" ] && kill -0 "$LAUNCHED_PID" 2>/dev/null; then + kill "$LAUNCHED_PID" 2>/dev/null || true + ABORT_WAIT_COUNT=0 + while kill -0 "$LAUNCHED_PID" 2>/dev/null && [ "$ABORT_WAIT_COUNT" -lt 20 ]; do + ABORT_WAIT_COUNT=$((ABORT_WAIT_COUNT + 1)) + sleep 0.1 + done + fi + if [ -n "$LAUNCHED_PID" ]; then + rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$BASE_DIR_FILE" + fi + exit 1 } trap cleanup_runner_next EXIT +trap abort_remote_launch HUP INT TERM +acquire_state_lock cat >"$RUNNER_NEXT" <<'SH' @@T3_RUNNER_SCRIPT@@ SH @@ -500,6 +574,76 @@ wait_for_pid_exit() { sleep 0.1 done } +discover_running_runtime() { + SERVICE_PID="" + if command -v systemctl >/dev/null 2>&1; then + SERVICE_PID="$(systemctl --user show t3code.service --property=MainPID --value 2>/dev/null || true)" + fi + node - "$DEFAULT_SERVER_HOME" "$BASE_DIR_FILE" "$DISCOVERED_BASE_DIR_FILE" "$SERVICE_PID" "\${T3CODE_HOME:-}" <<'NODE' +const fs = require("node:fs"); +const path = require("node:path"); +const defaultBaseDir = process.argv[2] ?? ""; +const knownBaseDirPath = process.argv[3] ?? ""; +const baseDirOutputPath = process.argv[4] ?? ""; +const servicePid = Number.parseInt(process.argv[5] ?? "", 10); +const environmentBaseDir = process.argv[6] ?? ""; +const candidates = []; +const seen = new Set(); +const addBaseDir = (value) => { + const baseDir = value.trim(); + if (baseDir.length === 0 || seen.has(baseDir)) return; + seen.add(baseDir); + candidates.push(baseDir); +}; +const addBaseDirFromPid = (pid) => { + if (!Number.isInteger(pid) || pid <= 0) return; + try { + const environment = fs.readFileSync(\`/proc/\${pid}/environ\`, "utf8").split("\\0"); + const value = environment.find((entry) => entry.startsWith("T3CODE_HOME=")); + if (value) addBaseDir(value.slice("T3CODE_HOME=".length)); + } catch {} +}; + +addBaseDir(environmentBaseDir); +try { + addBaseDir(fs.readFileSync(knownBaseDirPath, "utf8")); +} catch {} +addBaseDirFromPid(servicePid); +addBaseDir(defaultBaseDir); + +const isAlive = (pid) => { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return error && typeof error === "object" && error.code === "EPERM"; + } +}; + +for (const baseDir of candidates) { + for (const variant of ["userdata", "dev"]) { + try { + const runtime = JSON.parse( + fs.readFileSync(path.join(baseDir, variant, "server-runtime.json"), "utf8"), + ); + const pid = Number(runtime.pid); + const port = Number(runtime.port); + if (!Number.isInteger(pid) || pid <= 0 || !Number.isInteger(port) || !isAlive(pid)) { + continue; + } + const origin = new URL(String(runtime.origin ?? "")); + if (origin.protocol !== "http:" || !["127.0.0.1", "localhost"].includes(origin.hostname)) { + continue; + } + fs.writeFileSync(baseDirOutputPath, \`\${baseDir}\\n\`, { mode: 0o600 }); + process.stdout.write(String(port)); + process.exit(0); + } catch {} + } +} +process.exitCode = 1; +NODE +} resolve_default_runtime_port() { node - "$DEFAULT_RUNTIME_FILE" <<'NODE' const fs = require("node:fs"); @@ -525,17 +669,27 @@ NODE REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" REMOTE_PORT="$(cat "$PORT_FILE" 2>/dev/null || true)" REMOTE_MANAGED="$(cat "$MANAGED_FILE" 2>/dev/null || true)" +RUNNER_RUNTIME_PORT="$(discover_running_runtime 2>/dev/null || true)" DEFAULT_RUNTIME_INFO="$(resolve_default_runtime_port 2>/dev/null || true)" DEFAULT_RUNTIME_PID="" DEFAULT_REMOTE_PORT="" -if [ -n "$DEFAULT_RUNTIME_INFO" ]; then +if [ -n "$RUNNER_RUNTIME_PORT" ]; then + DEFAULT_REMOTE_PORT="$RUNNER_RUNTIME_PORT" +elif [ -n "$DEFAULT_RUNTIME_INFO" ]; then DEFAULT_RUNTIME_PID="\${DEFAULT_RUNTIME_INFO%% *}" DEFAULT_REMOTE_PORT="\${DEFAULT_RUNTIME_INFO#* }" + printf '%s\n' "$DEFAULT_SERVER_HOME" >"$DISCOVERED_BASE_DIR_FILE" fi if [ -n "$DEFAULT_REMOTE_PORT" ]; then + PREVIOUS_REMOTE_PORT="$REMOTE_PORT" REMOTE_PORT="$DEFAULT_REMOTE_PORT" if wait_ready "@@T3_REUSE_READY_TIMEOUT_MS@@"; then - if [ "$REMOTE_MANAGED" = "managed" ]; then + if [ -f "$DISCOVERED_BASE_DIR_FILE" ]; then + mv "$DISCOVERED_BASE_DIR_FILE" "$BASE_DIR_FILE" + fi + if [ "$REMOTE_MANAGED" = "managed" ] && [ "$PREVIOUS_REMOTE_PORT" = "$DEFAULT_REMOTE_PORT" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMOTE_PID" 2>/dev/null; then + printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" + elif [ "$REMOTE_MANAGED" = "managed" ]; then PID_TO_STOP="\${REMOTE_PID:-$DEFAULT_RUNTIME_PID}" if [ -n "$PID_TO_STOP" ] && kill -0 "$PID_TO_STOP" 2>/dev/null; then kill "$PID_TO_STOP" 2>/dev/null || true @@ -554,9 +708,11 @@ if [ -n "$DEFAULT_REMOTE_PORT" ]; then REMOTE_MANAGED="external" fi else - REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" - REMOTE_PORT="$(cat "$PORT_FILE" 2>/dev/null || true)" - REMOTE_MANAGED="$(cat "$MANAGED_FILE" 2>/dev/null || true)" + printf 'Existing remote T3 server did not become ready on 127.0.0.1:%s.\n' "$REMOTE_PORT" >&2 + REMOTE_PORT="$PREVIOUS_REMOTE_PORT" + DEFAULT_REMOTE_PORT="" + DEFAULT_RUNTIME_PID="" + rm -f "$DISCOVERED_BASE_DIR_FILE" fi fi if [ "$REMOTE_MANAGED" = "external" ]; then @@ -590,8 +746,10 @@ if [ -z "$REMOTE_PORT" ]; then printf 'Failed to find an available port on the remote host. Ensure node is available on PATH.\\n' >&2 exit 1 fi - nohup env T3CODE_NO_BROWSER=1 "$RUNNER_FILE" serve --host 127.0.0.1 --port "$REMOTE_PORT" --base-dir "$DEFAULT_SERVER_HOME" >>"$LOG_FILE" 2>&1 < /dev/null & + nohup env T3CODE_NO_BROWSER=1 "$RUNNER_FILE" serve --host 127.0.0.1 --port "$REMOTE_PORT" >>"$LOG_FILE" 2>&1 < /dev/null 9>&- & REMOTE_PID="$!" + LAUNCHED_PID="$REMOTE_PID" + printf '%s\\n' "\${T3CODE_HOME:-$DEFAULT_SERVER_HOME}" >"$BASE_DIR_FILE" printf '%s\\n' "$REMOTE_PID" >"$PID_FILE" printf '%s\\n' "$REMOTE_PORT" >"$PORT_FILE" printf 'managed\\n' >"$MANAGED_FILE" @@ -604,24 +762,70 @@ if [ -z "$REMOTE_PORT" ]; then fi kill "$REMOTE_PID" 2>/dev/null || true wait_for_pid_exit "$REMOTE_PID" - rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" + LAUNCHED_PID="" + rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$BASE_DIR_FILE" exit 1 fi + LAUNCHED_PID="" fi printf '{"remotePort":%s,"serverKind":"%s"}\\n' "$REMOTE_PORT" "\${REMOTE_MANAGED:-managed}" `; export const REMOTE_PAIRING_SCRIPT = `set -eu STATE_DIR="$HOME/.t3/ssh-launch/@@T3_STATE_KEY@@" -DEFAULT_SERVER_HOME="$HOME/.t3" RUNNER_FILE="$STATE_DIR/run-t3.sh" +RUNNER_NEXT="$STATE_DIR/run-t3.next.$$" +BASE_DIR_FILE="$STATE_DIR/base-dir" +PAIR_OUTPUT_FILE="$STATE_DIR/pair-output.$$" mkdir -p "$STATE_DIR" -cat >"$RUNNER_FILE" <<'SH' +umask 077 +@@T3_STATE_LOCK_SCRIPT@@ +cleanup_pairing_files() { + rm -f "$PAIR_OUTPUT_FILE" "$RUNNER_NEXT" + release_state_lock +} +trap cleanup_pairing_files EXIT +acquire_state_lock +cat >"$RUNNER_NEXT" <<'SH' @@T3_RUNNER_SCRIPT@@ SH -chmod 700 "$RUNNER_FILE" -PAIRING_BASE_DIR="$DEFAULT_SERVER_HOME" -"$RUNNER_FILE" auth pairing create --base-dir "$PAIRING_BASE_DIR" --json +chmod 700 "$RUNNER_NEXT" +mv -f "$RUNNER_NEXT" "$RUNNER_FILE" +PAIRING_BASE_DIR="$(cat "$BASE_DIR_FILE" 2>/dev/null || true)" +if [ -z "$PAIRING_BASE_DIR" ]; then + printf 'SSH environment state is missing its T3 base directory. Retry the connection.\n' >&2 + exit 1 +fi +PAIR_ATTEMPT=0 +PAIR_SUCCEEDED=0 +while [ "$PAIR_ATTEMPT" -lt 5 ]; do + PAIR_ATTEMPT=$((PAIR_ATTEMPT + 1)) + if "$RUNNER_FILE" pair --base-dir "$PAIRING_BASE_DIR" >"$PAIR_OUTPUT_FILE" 2>&1; then + PAIR_SUCCEEDED=1 + break + fi + if ! grep -q 'database is locked' "$PAIR_OUTPUT_FILE"; then + break + fi + sleep 0.2 +done +if [ "$PAIR_SUCCEEDED" -ne 1 ]; then + printf 'Remote T3 server could not issue an SSH pairing credential.\n' >&2 + exit 1 +fi +node - "$PAIR_OUTPUT_FILE" <<'NODE' +const fs = require("node:fs"); +const outputPath = process.argv[2] ?? ""; +try { + const output = fs.readFileSync(outputPath, "utf8"); + const tokenLine = output.split(/\\r?\\n/u).findLast((line) => line.startsWith("Token: ")); + const credential = tokenLine?.slice("Token: ".length).trim() ?? ""; + if (credential.length === 0) process.exit(1); + process.stdout.write(JSON.stringify({ credential }) + "\\n"); +} catch { + process.exit(1); +} +NODE `; export const REMOTE_STOP_SCRIPT = `set -eu @@ -629,9 +833,22 @@ STATE_DIR="$HOME/.t3/ssh-launch/@@T3_STATE_KEY@@" PID_FILE="$STATE_DIR/pid" PORT_FILE="$STATE_DIR/port" MANAGED_FILE="$STATE_DIR/managed" +BASE_DIR_FILE="$STATE_DIR/base-dir" +mkdir -p "$STATE_DIR" +umask 077 +@@T3_STATE_LOCK_SCRIPT@@ +cleanup_stop() { + release_state_lock +} +trap cleanup_stop EXIT +acquire_state_lock REMOTE_MANAGED="$(cat "$MANAGED_FILE" 2>/dev/null || true)" REMOTE_PID="$(cat "$PID_FILE" 2>/dev/null || true)" -if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMOTE_PID" 2>/dev/null; then +if [ "$REMOTE_MANAGED" = "external" ]; then + printf '{"stopped":true}\n' + exit 0 +fi +if [ -n "$REMOTE_PID" ] && kill -0 "$REMOTE_PID" 2>/dev/null; then kill "$REMOTE_PID" 2>/dev/null || true WAIT_COUNT=0 while kill -0 "$REMOTE_PID" 2>/dev/null && [ "$WAIT_COUNT" -lt 20 ]; do @@ -639,7 +856,7 @@ if [ "$REMOTE_MANAGED" != "external" ] && [ -n "$REMOTE_PID" ] && kill -0 "$REMO sleep 0.1 done fi -rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" +rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE" "$BASE_DIR_FILE" printf '{"stopped":true}\\n' `; @@ -678,6 +895,7 @@ export function buildRemoteLaunchScript(input?: RemoteT3RunnerOptions): string { T3_RUNNER_SCRIPT: stripTrailingNewlines(buildRemoteT3RunnerScript(input)), T3_PICK_PORT_SCRIPT: stripTrailingNewlines(REMOTE_PICK_PORT_SCRIPT), T3_WAIT_READY_SCRIPT: stripTrailingNewlines(REMOTE_WAIT_READY_SCRIPT), + T3_STATE_LOCK_SCRIPT: stripTrailingNewlines(REMOTE_STATE_LOCK_SCRIPT), T3_DEFAULT_REMOTE_PORT: String(DEFAULT_REMOTE_PORT), T3_REMOTE_PORT_SCAN_WINDOW: String(REMOTE_PORT_SCAN_WINDOW), T3_READY_TIMEOUT_MS: String(REMOTE_READY_TIMEOUT_MS), @@ -693,12 +911,14 @@ export function buildRemotePairingScript( return applyScriptPlaceholders(REMOTE_PAIRING_SCRIPT, { T3_STATE_KEY: remoteStateKey(target), T3_RUNNER_SCRIPT: stripTrailingNewlines(buildRemoteT3RunnerScript(input)), + T3_STATE_LOCK_SCRIPT: stripTrailingNewlines(REMOTE_STATE_LOCK_SCRIPT), }); } export function buildRemoteStopScript(target: DesktopSshEnvironmentTarget): string { return applyScriptPlaceholders(REMOTE_STOP_SCRIPT, { T3_STATE_KEY: remoteStateKey(target), + T3_STATE_LOCK_SCRIPT: stripTrailingNewlines(REMOTE_STATE_LOCK_SCRIPT), }); } @@ -1177,6 +1397,15 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma Deferred.Deferred >(); const authSecrets = new Map(); + const pairingSemaphores = new Map(); + + const pairingSemaphoreFor = (key: string) => { + const existing = pairingSemaphores.get(key); + if (existing !== undefined) return existing; + const semaphore = Semaphore.makeUnsafe(1); + pairingSemaphores.set(key, semaphore); + return semaphore; + }; const closeTunnelEntry = Effect.fn("ssh/tunnel.closeTunnelEntry")(function* ( entry: SshTunnelEntry, @@ -1457,7 +1686,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma remotePort: entry.remotePort, }); const readinessExit = yield* Effect.exit( - waitForHttpReady({ baseUrl: entry.httpBaseUrl, timeoutMs: 2_000 }), + waitForHttpReady({ baseUrl: entry.httpBaseUrl, timeoutMs: SSH_READY_TIMEOUT_MS }), ); if (Exit.isSuccess(readinessExit)) { yield* Effect.logDebug("ssh.environment.tunnel.reused", { @@ -1552,11 +1781,13 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma const entry = yield* ensureTunnelEntry(key, resolvedTarget, runner); const pairingResult = requestOptions?.issuePairingToken - ? yield* runWithSshAuth({ - key, - target: entry.target, - operation: (authOptions) => issueRemotePairingToken(entry.target, authOptions, runner), - }) + ? yield* pairingSemaphoreFor(key).withPermit( + runWithSshAuth({ + key, + target: entry.target, + operation: (authOptions) => issueRemotePairingToken(entry.target, authOptions, runner), + }), + ) : null; const pairingToken = pairingResult?.credential ?? null;