diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index ba2cf5c5ac05..19c5467a6684 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -1,12 +1,17 @@ +import { NodeServices } from "@effect/platform-node"; import { describe, expect, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; +import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PlatformError from "effect/PlatformError"; +import * as References from "effect/References"; +import * as Scope from "effect/Scope"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -14,6 +19,7 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import * as RelayClient from "@t3tools/shared/relayClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import { CLOUD_ENDPOINT_RUNTIME_CONFIG, encodeEndpointRuntimeConfigJson } from "./config.ts"; import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts"; const relayClientAvailableLayer = Layer.succeed( @@ -62,26 +68,192 @@ function makeHandle(input: { readonly onKill: () => void; readonly isRunning?: () => boolean; readonly exitCode?: Effect.Effect; + readonly all?: ChildProcessSpawner.ChildProcessHandle["all"]; }) { + const stopped = Deferred.makeUnsafe(); return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(input.pid), - exitCode: input.exitCode ?? Effect.never, + exitCode: (input.exitCode ?? Effect.never).pipe(Effect.race(Deferred.await(stopped))), isRunning: Effect.sync(() => input.isRunning?.() ?? true), kill: () => Effect.sync(() => { input.onKill(); - }), + }).pipe(Effect.andThen(Deferred.succeed(stopped, ChildProcessSpawner.ExitCode(0)))), unref: Effect.succeed(Effect.void), 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, }); } describe("CloudManagedEndpointRuntime", () => { + it.effect.each([ + { change: "replacement", maxOps: 64 }, + { change: "replacement", maxOps: 2048 }, + { change: "disable", maxOps: 64 }, + { change: "disable", maxOps: 2048 }, + ])( + "does not report a stopped connector after $change with scheduling budget $maxOps", + ({ change, maxOps }) => + Effect.gen(function* () { + const receipts: Array = []; + let count = 0; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const pid = 710 + count++; + const handle = makeHandle({ + pid, + isRunning: () => !receipts.includes(`killed-${pid}`), + onKill: () => receipts.push(`killed-${pid}`), + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const config = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "first", + tunnelId: "old-tunnel", + }; + expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 710 }); + const replacement = + change === "disable" + ? null + : { ...config, connectorToken: "replacement", tunnelId: "new-tunnel" }; + const statuses = yield* Effect.all( + [ + runtime + .applyConfig(config) + .pipe( + Effect.tap((status) => Effect.sync(() => receipts.push(`first-${status.status}`))), + ), + runtime + .applyConfig(replacement) + .pipe( + Effect.tap((status) => Effect.sync(() => receipts.push(`second-${status.status}`))), + ), + ], + { concurrency: 2 }, + ); + expect(statuses[1]).toMatchObject( + change === "disable" + ? { status: "disabled" } + : { status: "running", pid: 711, tunnelId: "new-tunnel" }, + ); + expect(statuses[0]).toMatchObject({ tunnelId: "old-tunnel" }); + expect(receipts).toContain("killed-710"); + if (statuses[0].status === "running") { + expect(statuses[0].pid).toBe(710); + expect(receipts.indexOf("first-running")).toBeLessThan(receipts.indexOf("killed-710")); + } else { + expect(statuses[0]).toMatchObject({ + status: "failed", + reason: "Relay client configuration changed before its connection could be confirmed.", + }); + } + // Default scheduling and a shorter operation budget exercise both request + // orderings without delaying process finalizers or sleeping in the fixture. + }).pipe(Effect.provideService(References.MaxOpsBeforeYield, maxOps)), + ); + it.effect("does not report registration as ready when the child is no longer running", () => + Effect.gen(function* () { + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ pid: 720, isRunning: () => false, onKill: () => {} }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + expect( + yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "stopped", + tunnelId: "stopped-tunnel", + }), + ).toEqual({ + status: "failed", + providerKind: "cloudflare_tunnel", + tunnelId: "stopped-tunnel", + reason: "Relay client is no longer running or its status could not be confirmed.", + }); + }), + ); + it.effect("a live owned child is not ready until its real output reports registration", () => + Effect.gen(function* () { + const nativeSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const spawned = yield* Deferred.make(); + const warningObserved = yield* Deferred.make(); + let spawnCount = 0; + const spawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + expect(ChildProcess.isStandardCommand(command)).toBe(true); + if (!ChildProcess.isStandardCommand(command)) + return yield* Effect.die("standard command required"); + expect(command.args).toEqual(["tunnel", "run"]); + expect(command.options.env?.TUNNEL_TOKEN).toBe("audit-synthetic-token"); + spawnCount += 1; + // This child owns no sockets and cannot contact a relay. Its real pipes + // model a running connector whose registration arrives after a timeout. + const child = yield* nativeSpawner.spawn( + ChildProcess.make( + process.execPath, + [ + "-e", + [ + "process.stdin.resume();", + "process.stdout.write('WRN transport connection unavailable\\n');", + "process.stdin.once('data', () => process.stdout.write('INF Registered tunnel connection connIndex=0\\n'));", + ].join("\n"), + ], + { stdin: "pipe", stdout: "pipe", stderr: "pipe" }, + ), + ); + yield* Deferred.succeed(spawned, child); + return ChildProcessSpawner.makeHandle({ + ...child, + all: child.all.pipe(Stream.tap(() => Deferred.succeed(warningObserved, undefined))), + }); + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const config = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "audit-synthetic-token", + tunnelId: "audit-owned-child", + }; + const applying = yield* runtime.applyConfig(config).pipe(Effect.forkChild); + const child = yield* Deferred.await(spawned); + yield* Deferred.await(warningObserved); + expect(yield* child.isRunning).toBe(true); + yield* TestClock.adjust("15 seconds"); + const beforeRegistration = yield* Fiber.join(applying); + expect(beforeRegistration).toMatchObject({ status: "failed", tunnelId: "audit-owned-child" }); + expect(yield* child.isRunning).toBe(true); + yield* Stream.make(new TextEncoder().encode("register\n")).pipe( + Stream.concat(Stream.never), + Stream.run(child.stdin), + Effect.forkChild, + ); + const registered = yield* runtime.applyConfig(config); + expect(registered).toMatchObject({ status: "running", pid: Number(child.pid) }); + expect(spawnCount).toBe(1); + expect(yield* child.isRunning).toBe(true); + expect(yield* runtime.applyConfig(null)).toEqual({ status: "disabled" }); + yield* Effect.result(child.exitCode); + expect(yield* child.isRunning).toBe(false); + }).pipe(Effect.provide(NodeServices.layer), Effect.orDie), + ); it("classifies Cloudflare connection and warning output", () => { expect( ManagedEndpointRuntime.classifyRelayClientOutput( @@ -405,6 +577,61 @@ describe("CloudManagedEndpointRuntime", () => { }).pipe(Effect.provide(TestClock.layer())), ); + 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 stopped = yield* runtime.applyConfig(null); + expect(stopped).toEqual({ status: "disabled" }); + expect(killed).toEqual([411, 412]); + }), + ); + it.effect("serializes concurrent connector config changes", () => Effect.gen(function* () { const spawned: Array = []; @@ -455,6 +682,488 @@ 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 outputStarted = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 600, + all: Stream.fromEffect( + Deferred.succeed(outputStarted, undefined).pipe( + Effect.andThen(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* Deferred.await(outputStarted); + + 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("clears a pending explicit config within the shutdown release deadline", () => + Effect.gen(function* () { + const outputStarted = yield* Deferred.make(); + const stoppedAt = yield* Deferred.make(); + const killed: Array = []; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 603, + all: Stream.fromEffect(Deferred.succeed(outputStarted, undefined)).pipe( + Stream.flatMap(() => Stream.never), + ), + onKill: () => { + killed.push(603); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const pending = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-pending", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(outputStarted); + const stoppingAt = yield* Clock.currentTimeMillis; + const stop = yield* runtime.applyConfig(null).pipe( + Effect.tap(() => + Clock.currentTimeMillis.pipe(Effect.flatMap((now) => Deferred.succeed(stoppedAt, now))), + ), + Effect.timeoutOption("10 seconds"), + Effect.forkChild, + ); + yield* TestClock.adjust("10 seconds"); + const stopped = yield* Fiber.join(stop); + yield* Fiber.interrupt(pending); + expect(stopped).toEqual(Option.some({ status: "disabled" })); + expect(yield* Deferred.await(stoppedAt)).toBe(stoppingAt); + expect(killed).toEqual([603]); + }), + ); + + 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("cancelling a reused registration wait does not take ownership of its child", () => + Effect.gen(function* () { + const registration = yield* Deferred.make(); + const outputStarted = yield* Deferred.make(); + const reused = yield* Deferred.make(); + const killed: Array = []; + let spawnCount = 0; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + spawnCount += 1; + const handle = makeHandle({ + pid: 610, + all: Stream.fromEffect(Deferred.succeed(outputStarted, undefined)).pipe( + Stream.flatMap(() => Stream.fromEffect(Deferred.await(registration))), + Stream.map(() => new TextEncoder().encode("INF Registered tunnel connection\n")), + ), + onKill: () => { + killed.push(610); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return ChildProcessSpawner.makeHandle({ + ...handle, + isRunning: Deferred.succeed(reused, undefined).pipe(Effect.as(true)), + }); + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const config = { providerKind: "cloudflare_tunnel" as const, connectorToken: "shared-token" }; + const owner = yield* runtime.applyConfig(config).pipe(Effect.forkChild); + yield* Deferred.await(outputStarted); + const reuser = yield* runtime.applyConfig(config).pipe(Effect.forkChild); + yield* Deferred.await(reused); + yield* Fiber.interrupt(reuser); + expect(killed).toEqual([]); + yield* Deferred.succeed(registration, undefined); + expect(yield* Fiber.join(owner)).toMatchObject({ status: "running", pid: 610 }); + expect(spawnCount).toBe(1); + }), + ); + + it.effect("cancelling an old registration wait never stops its replacement", () => + Effect.gen(function* () { + const outputStarted = yield* Deferred.make(); + const killed: Array = []; + let spawnCount = 0; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const pid = 620 + spawnCount++; + const handle = makeHandle({ + pid, + ...(pid === 620 + ? { + all: Stream.fromEffect(Deferred.succeed(outputStarted, undefined)).pipe( + Stream.flatMap(() => Stream.never), + ), + } + : {}), + onKill: () => { + killed.push(pid); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + // Hold the old wait's exit delivery to exercise cancellation after + // replacement, independently of the supervisor's scheduling. + return pid === 620 + ? ChildProcessSpawner.makeHandle({ ...handle, exitCode: Effect.never }) + : handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const previous = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "old-token", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(outputStarted); + const nextConfig = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "new-token", + }; + expect(yield* runtime.applyConfig(nextConfig)).toMatchObject({ status: "running", pid: 621 }); + yield* Fiber.interrupt(previous); + expect(killed).toEqual([620]); + expect(yield* runtime.applyConfig(nextConfig)).toMatchObject({ status: "running", pid: 621 }); + expect(spawnCount).toBe(2); + }), + ); + + it.effect("cancels a queued apply without waiting for another caller's spawn", () => + Effect.gen(function* () { + const spawning = yield* Deferred.make(); + const releaseSpawn = yield* Deferred.make(); + const queuedStarted = yield* Deferred.make(); + const killed: Array = []; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + yield* Deferred.succeed(spawning, undefined); + yield* Deferred.await(releaseSpawn); + const handle = makeHandle({ + pid: 630, + onKill: () => { + killed.push(630); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + const runtime = yield* buildCloudManagedEndpointRuntime(spawner); + const first = yield* runtime + .applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "first-token", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(spawning); + const queued = yield* Deferred.succeed(queuedStarted, undefined).pipe( + Effect.andThen(runtime.applyConfig(null)), + Effect.forkChild, + ); + yield* Deferred.await(queuedStarted); + yield* Fiber.interrupt(queued); + expect(killed).toEqual([]); + yield* Deferred.succeed(releaseSpawn, undefined); + expect(yield* Fiber.join(first)).toMatchObject({ status: "running", pid: 630 }); + }), + ); + + it.effect("builds the layer without waiting for a persisted config to register", () => + Effect.gen(function* () { + const killed: Array = []; + const spawned = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 700, + all: Stream.never, + onKill: () => { + killed.push(700); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + yield* Deferred.succeed(spawned, undefined); + return handle; + }), + ); + const configJson = yield* encodeEndpointRuntimeConfigJson({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }); + const secretStoreLayer = Layer.mock(ServerSecretStore.ServerSecretStore)({ + get: (name) => + Effect.succeed( + name === CLOUD_ENDPOINT_RUNTIME_CONFIG + ? Option.some(new TextEncoder().encode(configJson)) + : Option.none(), + ), + }); + + const scope = yield* Scope.make("sequential"); + yield* Layer.build( + ManagedEndpointRuntime.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + relayClientAvailableLayer, + secretStoreLayer, + ), + ), + ), + ).pipe(Effect.provideService(Scope.Scope, scope)); + yield* Deferred.await(spawned); + yield* Scope.close(scope, Exit.void); + + expect(killed).toEqual([700]); + }), + ); + + it.effect("stops the boot connector without waiting for it to register", () => + Effect.gen(function* () { + const killed: Array = []; + const spawned = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 701, + all: Stream.never, + onKill: () => { + killed.push(701); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + yield* Deferred.succeed(spawned, undefined); + return handle; + }), + ); + const configJson = yield* encodeEndpointRuntimeConfigJson({ + providerKind: "cloudflare_tunnel", + connectorToken: "token-secret", + tunnelId: "tunnel-1", + }); + const secretStoreLayer = Layer.mock(ServerSecretStore.ServerSecretStore)({ + get: (name) => + Effect.succeed( + name === CLOUD_ENDPOINT_RUNTIME_CONFIG + ? Option.some(new TextEncoder().encode(configJson)) + : Option.none(), + ), + }); + + const scope = yield* Scope.make("sequential"); + const context = yield* Layer.build( + ManagedEndpointRuntime.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + relayClientAvailableLayer, + secretStoreLayer, + ), + ), + ), + ).pipe(Effect.provideService(Scope.Scope, scope)); + const runtime = yield* Effect.service( + ManagedEndpointRuntime.CloudManagedEndpointRuntime, + ).pipe(Effect.provide(context)); + yield* Deferred.await(spawned); + + // The shutdown tunnel release runs this call under a 10 second timeout. + // It must not queue behind a boot apply that waits for registration. + const stopped = yield* runtime.applyConfig(null); + + expect(stopped).toEqual({ status: "disabled" }); + expect(killed).toEqual([701]); + yield* Scope.close(scope, Exit.void); + }), + ); + 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 564d7346d036..5e499e0934a0 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -2,9 +2,11 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay import * as RelayClient from "@t3tools/shared/relayClient"; import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; @@ -18,6 +20,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); } @@ -65,6 +69,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; @@ -116,23 +121,73 @@ export const make = Effect.gen(function* () { const relayClient = yield* RelayClient.RelayClient; const activeRef = yield* Ref.make(null); const desiredConfigRef = yield* Ref.make(null); + const configApplied = yield* Ref.make(false); const reconcileSemaphore = yield* Semaphore.make(1); const restartDelayRef = yield* Ref.make(0); - let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"]; + let startConnector: ( + config: RelayManagedEndpointRuntimeConfig, + configKey: string, + ) => Effect.Effect; const stopActive = Effect.gen(function* () { const active = yield* Ref.getAndSet(activeRef, null); yield* stopConnector(active); }); + const awaitConnectorConnection = Effect.fn( + "CloudManagedEndpointRuntime.awaitConnectorConnection", + )(function* (connector: ActiveConnector) { + const failed = (reason: string) => + ({ + status: "failed", + providerKind: "cloudflare_tunnel", + reason, + ...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}), + ...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}), + }) satisfies CloudManagedEndpointRuntimeStatus; + 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 yield* reconcileSemaphore.withPermits(1)( + Effect.gen(function* () { + if ((yield* Ref.get(activeRef)) !== connector) { + return failed( + "Relay client configuration changed before its connection could be confirmed.", + ); + } + const isRunning = yield* connector.child.isRunning.pipe( + Effect.orElseSucceed(() => false), + ); + if (!isRunning) { + return failed( + "Relay client is no longer running or its status could not be confirmed.", + ); + } + 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 failed(reason); + }); + const superviseConnector = (connector: ActiveConnector) => Effect.gen(function* () { const result = yield* Effect.result(connector.child.exitCode); const activeAtExit = yield* Ref.get(activeRef); - if ( - activeAtExit?.child.pid !== connector.child.pid || - activeAtExit.configKey !== connector.configKey - ) { + if (activeAtExit !== connector) { return; } const uptimeMillis = (yield* Clock.currentTimeMillis) - connector.startedAtMillis; @@ -164,10 +219,7 @@ export const make = Effect.gen(function* () { 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); @@ -190,7 +242,7 @@ export const make = Effect.gen(function* () { tunnelId: connector.config.tunnelId, tunnelName: connector.config.tunnelName, }); - yield* reconcileConfig(desiredConfig); + yield* startConnector(desiredConfig, connector.configKey); }), ); }).pipe( @@ -213,7 +265,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": @@ -230,7 +286,106 @@ export const make = Effect.gen(function* () { ), ); - reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) { + 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, + startedAtMillis: yield* Clock.currentTimeMillis, + } 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; + }), + ); + }, + ); + + const reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* ( + config: RelayManagedEndpointRuntimeConfig | null, + ): Effect.fn.Return { if (!config || config.providerKind !== "cloudflare_tunnel") { yield* stopActive; return config @@ -243,116 +398,78 @@ 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 active; } } yield* stopActive; + return yield* startConnector(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; - } - - 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, + const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(function* ( + config: RelayManagedEndpointRuntimeConfig | null, + ) { + let started: ActiveConnector | null = null; + return yield* reconcileSemaphore + .withPermits(1)( + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + // An explicit config change starts over with a fresh backoff. + yield* Ref.set(restartDelayRef, 0); + yield* Ref.set(configApplied, true); + yield* Ref.set(desiredConfigRef, config); + const previous = yield* Ref.get(activeRef); + const result = yield* restore(reconcileConfig(config)).pipe( + Effect.onInterrupt(() => + Effect.gen(function* () { + if ((yield* Ref.get(activeRef)) !== previous) yield* stopActive; + }), + ), + ); + if ("child" in result && result !== previous) started = result; + return result; }), ), - 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), - ), + ) + .pipe( + // Only process ownership is serialized. Registration must not delay + // another config or the shorter shutdown tunnel-release deadline. + Effect.flatMap((result) => + "child" in result ? awaitConnectorConnection(result) : Effect.succeed(result), ), + Effect.onInterrupt(() => { + const owned = started; + if (owned === null) return Effect.void; + return reconcileSemaphore.withPermits(1)( + Effect.gen(function* () { + // A reused child belongs to the earlier apply. A replacement may + // already belong to a later one; neither is ours to cancel. + if ((yield* Ref.get(activeRef)) === owned) yield* stopActive; + }), + ); + }), ); - - if ("status" in child && child.status === "failed") { - return child; - } - - if (!("status" in child)) { - const connector = { - child, - scope: connectorScope, - configKey: nextConfigKey, - config, - startedAtMillis: yield* Clock.currentTimeMillis, - } 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; }); - const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")( - (config: RelayManagedEndpointRuntimeConfig | null) => - reconcileSemaphore.withPermits(1)( - // An explicit config change starts over with a fresh backoff. - Ref.set(restartDelayRef, 0).pipe( - Effect.andThen(Ref.set(desiredConfigRef, config)), - Effect.andThen(reconcileConfig(config)), - ), - ), - ); + // The boot apply discards its status, so it must not hold the reconcile + // permit through the registration wait. The shutdown tunnel release calls + // applyConfig(null) under a shorter timeout and would expire behind it. + const applyInitialConfig = (config: RelayManagedEndpointRuntimeConfig | null) => + reconcileSemaphore.withPermits(1)( + Effect.gen(function* () { + // A caller can apply a config while the boot apply still waits for the + // permit. That newer config wins. The boot config must not replace it. + if (yield* Ref.get(configApplied)) { + return; + } + yield* Ref.set(desiredConfigRef, config); + if (!config || config.providerKind !== "cloudflare_tunnel") { + yield* stopActive; + return; + } + yield* startConnector(config, runtimeConfigKey(config)); + }), + ); const runtime = CloudManagedEndpointRuntime.of({ applyConfig, @@ -365,8 +482,12 @@ export const make = Effect.gen(function* () { ), ), ); - yield* runtime.applyConfig(initialConfig); - yield* Effect.addFinalizer(() => runtime.applyConfig(null)); + // Interrupt a boot apply that still starts the connector, so the final + // reconcile does not wait for a spawn it will stop right after. + const startup = yield* applyInitialConfig(initialConfig).pipe(Effect.forkScoped); + yield* Effect.addFinalizer(() => + Fiber.interrupt(startup).pipe(Effect.andThen(runtime.applyConfig(null))), + ); return runtime; }); diff --git a/apps/server/src/cloud/http.ts b/apps/server/src/cloud/http.ts index e0d458b4b97c..4e700432016e 100644 --- a/apps/server/src/cloud/http.ts +++ b/apps/server/src/cloud/http.ts @@ -468,7 +468,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 bebde7eded76..62f69cf6278a 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -3614,17 +3614,32 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); - it.effect("fails relay config when the managed endpoint connector cannot start", () => + it.effect.each([ + "cloudflared missing", + "Relay client did not register a tunnel connection within 15 seconds. Check whether the network allows outbound TCP and UDP traffic on port 7844.", + ])("fails relay config when the managed endpoint connector cannot start: %s", (reason) => Effect.gen(function* () { + const applied: Array = []; + let ready = false; yield* buildAppUnderTest({ layers: { cloudManagedEndpointRuntime: { - applyConfig: () => - Effect.succeed({ - status: "failed", - providerKind: "cloudflare_tunnel", - reason: "cloudflared missing", - tunnelId: "tunnel-1", + applyConfig: (config) => + Effect.sync(() => { + applied.push(config); + return ready + ? { + status: "running", + providerKind: "cloudflare_tunnel", + pid: 100, + tunnelId: "tunnel-1", + } + : { + status: "failed", + providerKind: "cloudflare_tunnel", + reason, + tunnelId: "tunnel-1", + }; }), }, }, @@ -3636,23 +3651,24 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }); const ownerCookie = yield* getAuthenticatedSessionCookieHeader(); const relayConfigUrl = yield* getHttpServerUrl("/api/connect/relay-config"); + const payload = { + relayUrl: "https://relay.example.test", + cloudUserId: "user_123", + environmentCredential: "t3env_test_credential", + cloudMintPublicKey: cloudKeyPair.publicKey, + endpointRuntime: { + providerKind: "cloudflare_tunnel", + connectorToken: "connector-token", + tunnelId: "tunnel-1", + }, + }; const relayConfigResponse = yield* fetchEffect(relayConfigUrl, { method: "POST", headers: { cookie: ownerCookie, "content-type": "application/json", }, - body: jsonRequestBody({ - relayUrl: "https://relay.example.test", - cloudUserId: "user_123", - environmentCredential: "t3env_test_credential", - cloudMintPublicKey: cloudKeyPair.publicKey, - endpointRuntime: { - providerKind: "cloudflare_tunnel", - connectorToken: "connector-token", - tunnelId: "tunnel-1", - }, - }), + body: jsonRequestBody(payload), }); assert.equal(relayConfigResponse.status, 503); @@ -3662,9 +3678,13 @@ 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. ${reason}`, + ); assert.equal(relayConfigBody.endpointRuntimeStatus?.status, "failed"); - assert.equal(relayConfigBody.endpointRuntimeStatus?.reason, "cloudflared missing"); + assert.equal(relayConfigBody.endpointRuntimeStatus?.reason, reason); + assert.deepEqual(applied, [payload.endpointRuntime]); const now = yield* DateTime.now; const healthRequest = makeCloudEnvironmentHealthRequest({ @@ -3692,6 +3712,36 @@ it.layer(NodeServices.layer)("server router seam", (it) => { healthBody.message, "Cloud mint public key is not installed for this environment.", ); + + // A timeout does not trigger an implicit stop or persist a partial link. + // A successful retry installs credentials using the existing wire shape. + ready = true; + const retryResponse = yield* fetchEffect(relayConfigUrl, { + method: "POST", + headers: { cookie: ownerCookie, "content-type": "application/json" }, + body: jsonRequestBody(payload), + }); + assert.equal(retryResponse.status, 200); + assert.deepEqual(yield* responseJsonEffect(retryResponse), { + ok: true, + endpointRuntimeStatus: { + status: "running", + providerKind: "cloudflare_tunnel", + pid: 100, + tunnelId: "tunnel-1", + }, + }); + assert.deepEqual(applied, [payload.endpointRuntime, payload.endpointRuntime]); + const linkStateUrl = yield* getHttpServerUrl("/api/connect/link-state"); + const linkStateResponse = yield* fetchEffect(linkStateUrl, { + headers: { cookie: ownerCookie }, + }); + const linkState = yield* responseJsonEffect<{ + linked: boolean; + managedTunnelActive: boolean; + }>(linkStateResponse); + assert.equal(linkState.linked, true); + assert.equal(linkState.managedTunnelActive, true); }).pipe(Effect.provide(NodeHttpServer.layerTest)), );