diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index b45b5099252a..9281d75d0bea 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -8,6 +8,7 @@ import * as Option from "effect/Option"; import * as PlatformError from "effect/PlatformError"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -60,6 +61,7 @@ function makeHandle(input: { readonly onKill: () => void; readonly isRunning?: () => boolean; readonly exitCode?: Effect.Effect; + readonly all?: ChildProcessSpawner.ChildProcessHandle["all"]; }) { return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(input.pid), @@ -73,7 +75,13 @@ function makeHandle(input: { stdin: Sink.drain, stdout: Stream.empty, stderr: Stream.empty, - all: Stream.empty, + all: + input.all ?? + Stream.make( + new TextEncoder().encode( + "2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n", + ), + ), getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, }); @@ -283,6 +291,68 @@ describe("CloudManagedEndpointRuntime", () => { }), ); + it.effect("does not block config changes while a restarted connector registers", () => + Effect.gen(function* () { + const killed: Array = []; + const firstExit = yield* Deferred.make(); + const secondSpawned = yield* Deferred.make(); + const secondRegistration = yield* Deferred.make(); + let spawnCount = 0; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + spawnCount += 1; + const pid = 410 + spawnCount; + if (spawnCount === 2) { + yield* Deferred.succeed(secondSpawned, undefined); + } + const handle = makeHandle({ + pid, + ...(spawnCount === 1 + ? {} + : { + all: Stream.fromEffect(Deferred.await(secondRegistration)).pipe( + Stream.map(() => + new TextEncoder().encode( + "2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n", + ), + ), + ), + }), + exitCode: + spawnCount === 1 + ? Deferred.await(firstExit) + : (Effect.never as Effect.Effect), + onKill: () => { + killed.push(pid); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + + yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token", + tunnelId: "tunnel-1", + }); + yield* Deferred.succeed(firstExit, ChildProcessSpawner.ExitCode(1)); + yield* Deferred.await(secondSpawned); + + const stopFiber = yield* runtime.applyConfig(null).pipe(Effect.forkChild); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + const stopped = stopFiber.pollUnsafe(); + const killedAfterStop = [...killed]; + yield* Deferred.succeed(secondRegistration, undefined); + yield* Fiber.join(stopFiber); + + expect(stopped).toBeDefined(); + expect(killedAfterStop).toEqual([411, 412]); + }), + ); + it.effect("serializes concurrent connector config changes", () => Effect.gen(function* () { const spawned: Array = []; @@ -333,6 +403,198 @@ describe("CloudManagedEndpointRuntime", () => { }), ); + it.effect("does not report a running connector before Cloudflare registers it", () => + Effect.gen(function* () { + const killed: Array = []; + const registerConnection = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 600, + all: Stream.fromEffect(Deferred.await(registerConnection)).pipe( + Stream.map(() => + new TextEncoder().encode( + "2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n", + ), + ), + Stream.concat(Stream.never), + ), + onKill: () => { + killed.push(600); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + + const statusFiber = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }) + .pipe(Effect.forkChild); + yield* Effect.yieldNow; + + expect(statusFiber.pollUnsafe()).toBeUndefined(); + + yield* TestClock.adjust("15 seconds"); + const status = yield* Fiber.join(statusFiber); + + expect(status).toMatchObject({ + status: "failed", + providerKind: "cloudflare_tunnel", + reason: + "Relay client did not register a tunnel connection within 15 seconds. Check whether the network allows outbound TCP and UDP traffic on port 7844.", + tunnelId: "tunnel-1", + }); + expect(killed).toEqual([]); + + yield* Deferred.succeed(registerConnection, undefined); + const recovered = yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }); + + expect(recovered).toMatchObject({ + status: "running", + pid: 600, + tunnelId: "tunnel-1", + }); + expect(killed).toEqual([]); + }), + ); + + it.effect("restarts a connector that exits before Cloudflare registers it", () => + Effect.gen(function* () { + const killed: Array = []; + let spawnCount = 0; + const secondSpawned = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + spawnCount += 1; + const pid = 600 + spawnCount; + const handle = makeHandle({ + pid, + ...(spawnCount === 1 + ? { + all: Stream.never, + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + } + : {}), + onKill: () => { + killed.push(pid); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + if (spawnCount === 2) { + yield* Deferred.succeed(secondSpawned, undefined); + } + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + + const status = yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }); + + expect(status).toMatchObject({ + status: "failed", + providerKind: "cloudflare_tunnel", + reason: "Relay client exited before it registered a tunnel connection.", + tunnelId: "tunnel-1", + }); + yield* Deferred.await(secondSpawned); + const recovered = yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }); + + expect(recovered).toMatchObject({ + status: "running", + providerKind: "cloudflare_tunnel", + pid: 602, + tunnelId: "tunnel-1", + }); + expect(killed).toEqual([601]); + }), + ); + + it.effect("stops a connector when its first configuration is interrupted during spawn", () => + Effect.gen(function* () { + const killed: Array = []; + const processStarted = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 602, + onKill: () => { + killed.push(602); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + yield* Deferred.succeed(processStarted, undefined); + return yield* Effect.never; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + + const statusFiber = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(processStarted); + yield* Fiber.interrupt(statusFiber); + + expect(killed).toEqual([602]); + }), + ); + + it.effect("stops a connector when its first configuration is interrupted", () => + Effect.gen(function* () { + const killed: Array = []; + const outputStarted = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 602, + all: Stream.fromEffect(Deferred.succeed(outputStarted, undefined)).pipe( + Stream.flatMap(() => Stream.never), + ), + onKill: () => { + killed.push(602); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + + const statusFiber = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(outputStarted); + yield* Fiber.interrupt(statusFiber); + + expect(killed).toEqual([602]); + }), + ); + it.effect("reports connector spawn failures", () => Effect.gen(function* () { const spawner = ChildProcessSpawner.make(() => diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index 89c0a23783c0..73e9fbde095b 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -1,6 +1,7 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay"; import * as RelayClient from "@t3tools/shared/relayClient"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; @@ -16,6 +17,8 @@ import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawne import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import { CLOUD_ENDPOINT_RUNTIME_CONFIG, decodeRuntimeConfig } from "./config.ts"; +const RELAY_CONNECTION_TIMEOUT = "15 seconds"; + function bytesToString(bytes: Uint8Array): string { return new TextDecoder().decode(bytes); } @@ -63,6 +66,7 @@ export class CloudManagedEndpointRuntime extends Context.Service< interface ActiveConnector { readonly child: ChildProcessSpawner.ChildProcessHandle; + readonly connected: Deferred.Deferred; readonly scope: Scope.Closeable; readonly configKey: string; readonly config: RelayManagedEndpointRuntimeConfig; @@ -105,6 +109,10 @@ export const make = Effect.gen(function* () { const activeRef = yield* Ref.make(null); const desiredConfigRef = yield* Ref.make(null); const reconcileSemaphore = yield* Semaphore.make(1); + let startConnector: ( + config: RelayManagedEndpointRuntimeConfig, + configKey: string, + ) => Effect.Effect; let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"]; const stopActive = Effect.gen(function* () { @@ -112,16 +120,43 @@ export const make = Effect.gen(function* () { yield* stopConnector(active); }); + const awaitConnectorConnection = Effect.fn( + "CloudManagedEndpointRuntime.awaitConnectorConnection", + )(function* (connector: ActiveConnector) { + const outcome = yield* Deferred.await(connector.connected).pipe( + Effect.as("connected" as const), + Effect.race(Effect.result(connector.child.exitCode).pipe(Effect.as("exited" as const))), + Effect.timeoutOption(RELAY_CONNECTION_TIMEOUT), + ); + if (Option.isSome(outcome) && outcome.value === "connected") { + return { + status: "running", + providerKind: "cloudflare_tunnel", + pid: Number(connector.child.pid), + ...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}), + ...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}), + } satisfies CloudManagedEndpointRuntimeStatus; + } + + const reason = Option.isSome(outcome) + ? "Relay client exited before it registered a tunnel connection." + : `Relay client did not register a tunnel connection within ${RELAY_CONNECTION_TIMEOUT}. Check whether the network allows outbound TCP and UDP traffic on port 7844.`; + return { + status: "failed", + providerKind: "cloudflare_tunnel", + reason, + ...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}), + ...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}), + } satisfies CloudManagedEndpointRuntimeStatus; + }); + const superviseConnector = (connector: ActiveConnector) => Effect.gen(function* () { const result = yield* Effect.result(connector.child.exitCode); yield* reconcileSemaphore.withPermits(1)( Effect.gen(function* () { const active = yield* Ref.get(activeRef); - if ( - active?.child.pid !== connector.child.pid || - active.configKey !== connector.configKey - ) { + if (active !== connector) { return; } yield* Ref.set(activeRef, null); @@ -144,7 +179,7 @@ export const make = Effect.gen(function* () { tunnelId: connector.config.tunnelId, tunnelName: connector.config.tunnelName, }); - yield* reconcileConfig(desiredConfig); + yield* startConnector(desiredConfig, connector.configKey); }), ); }).pipe( @@ -167,7 +202,11 @@ export const make = Effect.gen(function* () { }; switch (classifyRelayClientOutput(line)) { case "connected": - return Effect.logInfo("Relay client tunnel connection registered", attributes); + return Deferred.succeed(connector.connected, undefined).pipe( + Effect.andThen( + Effect.logInfo("Relay client tunnel connection registered", attributes), + ), + ); case "warning": return Effect.logWarning("Relay client reported a transport warning", attributes); case "debug": @@ -184,6 +223,102 @@ export const make = Effect.gen(function* () { ), ); + startConnector = Effect.fn("CloudManagedEndpointRuntime.startConnector")( + function* (config, nextConfigKey) { + const executable = yield* relayClient.resolve; + if (executable.status !== "available") { + return { + status: "failed", + providerKind: "cloudflare_tunnel", + reason: + executable.status === "unsupported" + ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.` + : "The relay client is not installed.", + ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), + ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), + } satisfies CloudManagedEndpointRuntimeStatus; + } + + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const connectorScope = yield* Scope.make("sequential"); + const child = yield* restore( + spawner + .spawn( + ChildProcess.make(executable.executablePath, ["tunnel", "run"], { + detached: false, + env: { + ...process.env, + TUNNEL_TOKEN: config.connectorToken, + }, + shell: false, + stderr: "pipe", + stdout: "pipe", + }), + ) + .pipe( + Effect.provideService(Scope.Scope, connectorScope), + Effect.tap((child) => + Effect.logInfo("Relay client process started; waiting for tunnel connection", { + pid: Number(child.pid), + tunnelId: config.tunnelId, + tunnelName: config.tunnelName, + }), + ), + Effect.onInterrupt(() => + Scope.close(connectorScope, Exit.void).pipe(Effect.ignore), + ), + ), + ).pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to start relay client", { + cause, + tunnelId: config.tunnelId, + tunnelName: config.tunnelName, + }).pipe( + Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)), + Effect.as({ + status: "failed", + providerKind: "cloudflare_tunnel", + reason: String(cause), + ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), + ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), + } satisfies CloudManagedEndpointRuntimeStatus), + ), + ), + ); + + if ("status" in child && child.status === "failed") { + return child; + } + + if (!("status" in child)) { + const connected = yield* Deferred.make(); + const connector = { + child, + connected, + scope: connectorScope, + configKey: nextConfigKey, + config, + } satisfies ActiveConnector; + yield* Ref.set(activeRef, connector); + yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); + yield* Effect.forkIn(superviseConnector(connector), connectorScope); + return connector; + } + + return { + status: "failed", + providerKind: "cloudflare_tunnel", + reason: "Relay client did not start.", + ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), + ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), + } satisfies CloudManagedEndpointRuntimeStatus; + }), + ); + }, + ); + reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) { if (!config || config.providerKind !== "cloudflare_tunnel") { yield* stopActive; @@ -197,103 +332,17 @@ export const make = Effect.gen(function* () { if (active?.configKey === nextConfigKey) { const isRunning = yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false)); if (isRunning) { - return { - status: "running", - providerKind: "cloudflare_tunnel", - pid: Number(active.child.pid), - ...(active.config.tunnelId ? { tunnelId: active.config.tunnelId } : {}), - ...(active.config.tunnelName ? { tunnelName: active.config.tunnelName } : {}), - } satisfies CloudManagedEndpointRuntimeStatus; + return yield* awaitConnectorConnection(active); } } yield* stopActive; - - const executable = yield* relayClient.resolve; - if (executable.status !== "available") { - return { - status: "failed", - providerKind: "cloudflare_tunnel", - reason: - executable.status === "unsupported" - ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.` - : "The relay client is not installed.", - ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), - ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), - } satisfies CloudManagedEndpointRuntimeStatus; - } - - const connectorScope = yield* Scope.make("sequential"); - const child = yield* spawner - .spawn( - ChildProcess.make(executable.executablePath, ["tunnel", "run"], { - detached: false, - env: { - ...process.env, - TUNNEL_TOKEN: config.connectorToken, - }, - shell: false, - stderr: "pipe", - stdout: "pipe", - }), - ) - .pipe( - Effect.provideService(Scope.Scope, connectorScope), - Effect.tap((child) => - Effect.logInfo("Relay client process started; waiting for tunnel connection", { - pid: Number(child.pid), - tunnelId: config.tunnelId, - tunnelName: config.tunnelName, - }), - ), - Effect.catch((cause) => - Effect.logWarning("Failed to start relay client", { - cause, - tunnelId: config.tunnelId, - tunnelName: config.tunnelName, - }).pipe( - Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)), - Effect.as({ - status: "failed", - providerKind: "cloudflare_tunnel", - reason: String(cause), - ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), - ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), - } satisfies CloudManagedEndpointRuntimeStatus), - ), - ), - ); - - if ("status" in child && child.status === "failed") { - return child; - } - - if (!("status" in child)) { - const connector = { - child, - scope: connectorScope, - configKey: nextConfigKey, - config, - } satisfies ActiveConnector; - yield* Ref.set(activeRef, connector); - yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); - yield* Effect.forkIn(superviseConnector(connector), connectorScope); - return { - status: "running", - providerKind: "cloudflare_tunnel", - pid: Number(child.pid), - ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), - ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), - } satisfies CloudManagedEndpointRuntimeStatus; - } - - return { - status: "failed", - providerKind: "cloudflare_tunnel", - reason: "Relay client did not start.", - ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}), - ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}), - } satisfies CloudManagedEndpointRuntimeStatus; + return yield* startConnector(config, nextConfigKey).pipe( + Effect.flatMap((result) => + "child" in result ? awaitConnectorConnection(result) : Effect.succeed(result), + ), + Effect.onInterrupt(() => stopActive), + ); }); const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")( diff --git a/apps/server/src/cloud/http.ts b/apps/server/src/cloud/http.ts index 29fdfe8ece2f..b0b6c661935c 100644 --- a/apps/server/src/cloud/http.ts +++ b/apps/server/src/cloud/http.ts @@ -467,7 +467,10 @@ const applyCloudRelayConfig = Effect.fn("environment.cloud.applyRelayConfig")(fu endpointRuntimeStatus.status === "disabled" || endpointRuntimeStatus.status === "running"; if (!ok) { return yield* new EnvironmentCloudEndpointUnavailableError({ - message: "Managed endpoint runtime could not be started.", + message: + endpointRuntimeStatus.status === "failed" + ? `Managed endpoint runtime could not connect. ${endpointRuntimeStatus.reason}` + : "Managed endpoint runtime could not connect.", endpointRuntimeStatus, }); } diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index a9a2c3fa10d6..c4b2d6c57dca 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -2962,7 +2962,10 @@ it.layer(NodeServices.layer)("server router seam", (it) => { endpointRuntimeStatus?: { status?: string; reason?: string }; }>(relayConfigResponse); assert.equal(relayConfigBody._tag, "EnvironmentCloudEndpointUnavailableError"); - assert.equal(relayConfigBody.message, "Managed endpoint runtime could not be started."); + assert.equal( + relayConfigBody.message, + "Managed endpoint runtime could not connect. cloudflared missing", + ); assert.equal(relayConfigBody.endpointRuntimeStatus?.status, "failed"); assert.equal(relayConfigBody.endpointRuntimeStatus?.reason, "cloudflared missing");