From 8b11a27aabe315d173ca5c2c0216c7379c946849 Mon Sep 17 00:00:00 2001 From: "Mateo M." Date: Thu, 27 Aug 2026 05:31:16 +0200 Subject: [PATCH 1/5] fix(server): wait for managed tunnel registration --- .../src/cloud/ManagedEndpointRuntime.test.ts | 92 ++++++++++++++++++- .../src/cloud/ManagedEndpointRuntime.ts | 69 ++++++++++---- apps/server/src/cloud/http.ts | 5 +- apps/server/src/server.test.ts | 5 +- 4 files changed, 152 insertions(+), 19 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index b45b5099252a..095843d44919 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, }); @@ -333,6 +341,88 @@ describe("CloudManagedEndpointRuntime", () => { }), ); + it.effect("does not report a running connector before Cloudflare registers it", () => + Effect.gen(function* () { + const killed: Array = []; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 600, + all: Stream.make( + new TextEncoder().encode( + "2026-08-27T10:00:00Z WRN Failed to dial edge with token-secret\n", + ), + ).pipe(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. Last warning: 2026-08-27T10:00:00Z WRN Failed to dial edge with ", + tunnelId: "tunnel-1", + }); + expect(killed).toEqual([600]); + }), + ); + + it.effect("reports a connector that exits before Cloudflare registers it", () => + Effect.gen(function* () { + const killed: Array = []; + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const handle = makeHandle({ + pid: 601, + all: Stream.never, + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + onKill: () => { + killed.push(601); + }, + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + 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", + }); + expect(killed).toEqual([601]); + }), + ); + 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..6e4fa9c4e147 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,8 @@ export class CloudManagedEndpointRuntime extends Context.Service< interface ActiveConnector { readonly child: ChildProcessSpawner.ChildProcessHandle; + readonly connected: Deferred.Deferred; + readonly lastWarning: Ref.Ref; readonly scope: Scope.Closeable; readonly configKey: string; readonly config: RelayManagedEndpointRuntimeConfig; @@ -112,6 +117,38 @@ 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 lastWarning = yield* Ref.get(connector.lastWarning); + yield* stopActive; + 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}.`; + return { + status: "failed", + providerKind: "cloudflare_tunnel", + reason: lastWarning ? `${reason} Last warning: ${lastWarning}` : 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); @@ -167,9 +204,17 @@ 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); + return Ref.set(connector.lastWarning, output).pipe( + Effect.andThen( + Effect.logWarning("Relay client reported a transport warning", attributes), + ), + ); case "debug": return Effect.logDebug("Relay client output", attributes); } @@ -197,13 +242,7 @@ 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); } } @@ -269,8 +308,12 @@ export const make = Effect.gen(function* () { } if (!("status" in child)) { + const connected = yield* Deferred.make(); + const lastWarning = yield* Ref.make(null); const connector = { child, + connected, + lastWarning, scope: connectorScope, configKey: nextConfigKey, config, @@ -278,13 +321,7 @@ export const make = Effect.gen(function* () { 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 yield* awaitConnectorConnection(connector); } return { 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"); From a2c3314e346c3cfbbecbd12d2962f004273511d5 Mon Sep 17 00:00:00 2001 From: "Mateo M." Date: Thu, 27 Aug 2026 05:35:19 +0200 Subject: [PATCH 2/5] fix(server): keep relay diagnostics bounded --- .../src/cloud/ManagedEndpointRuntime.test.ts | 2 +- apps/server/src/cloud/ManagedEndpointRuntime.ts | 14 +++----------- 2 files changed, 4 insertions(+), 12 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 095843d44919..89c9f0407b28 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -381,7 +381,7 @@ describe("CloudManagedEndpointRuntime", () => { status: "failed", providerKind: "cloudflare_tunnel", reason: - "Relay client did not register a tunnel connection within 15 seconds. Last warning: 2026-08-27T10:00:00Z WRN Failed to dial edge with ", + "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([600]); diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index 6e4fa9c4e147..1fcdfefeceb9 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -67,7 +67,6 @@ export class CloudManagedEndpointRuntime extends Context.Service< interface ActiveConnector { readonly child: ChildProcessSpawner.ChildProcessHandle; readonly connected: Deferred.Deferred; - readonly lastWarning: Ref.Ref; readonly scope: Scope.Closeable; readonly configKey: string; readonly config: RelayManagedEndpointRuntimeConfig; @@ -135,15 +134,14 @@ export const make = Effect.gen(function* () { } satisfies CloudManagedEndpointRuntimeStatus; } - const lastWarning = yield* Ref.get(connector.lastWarning); yield* stopActive; 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}.`; + : `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: lastWarning ? `${reason} Last warning: ${lastWarning}` : reason, + reason, ...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}), ...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}), } satisfies CloudManagedEndpointRuntimeStatus; @@ -210,11 +208,7 @@ export const make = Effect.gen(function* () { ), ); case "warning": - return Ref.set(connector.lastWarning, output).pipe( - Effect.andThen( - Effect.logWarning("Relay client reported a transport warning", attributes), - ), - ); + return Effect.logWarning("Relay client reported a transport warning", attributes); case "debug": return Effect.logDebug("Relay client output", attributes); } @@ -309,11 +303,9 @@ export const make = Effect.gen(function* () { if (!("status" in child)) { const connected = yield* Deferred.make(); - const lastWarning = yield* Ref.make(null); const connector = { child, connected, - lastWarning, scope: connectorScope, configKey: nextConfigKey, config, From b2cb86610fb2b3456d60d086407836f619a571ba Mon Sep 17 00:00:00 2001 From: "Mateo M." Date: Thu, 27 Aug 2026 05:39:51 +0200 Subject: [PATCH 3/5] fix(server): stop interrupted tunnel startup --- .../src/cloud/ManagedEndpointRuntime.test.ts | 35 +++++++++++++++++++ .../src/cloud/ManagedEndpointRuntime.ts | 2 +- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 89c9f0407b28..42091e6df006 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -423,6 +423,41 @@ describe("CloudManagedEndpointRuntime", () => { }), ); + 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 1fcdfefeceb9..4cf50b4ccb4d 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -313,7 +313,7 @@ export const make = Effect.gen(function* () { yield* Ref.set(activeRef, connector); yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); yield* Effect.forkIn(superviseConnector(connector), connectorScope); - return yield* awaitConnectorConnection(connector); + return yield* awaitConnectorConnection(connector).pipe(Effect.onInterrupt(() => stopActive)); } return { From 78b9eb4eada1799b09b53306e628d55ed66ff67d Mon Sep 17 00:00:00 2001 From: "Mateo M." Date: Thu, 27 Aug 2026 09:49:10 +0200 Subject: [PATCH 4/5] fix(server): preserve relay connector recovery --- .../src/cloud/ManagedEndpointRuntime.test.ts | 95 ++++++++++-- .../src/cloud/ManagedEndpointRuntime.ts | 137 ++++++++++-------- 2 files changed, 158 insertions(+), 74 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 42091e6df006..4de6c7cb53cb 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -344,15 +344,19 @@ 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.make( - new TextEncoder().encode( - "2026-08-27T10:00:00Z WRN Failed to dial edge with token-secret\n", + all: Stream.fromEffect(Deferred.await(registerConnection)).pipe( + Stream.map(() => + new TextEncoder().encode( + "2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n", + ), ), - ).pipe(Stream.concat(Stream.never)), + Stream.concat(Stream.never), + ), onKill: () => { killed.push(600); }, @@ -384,24 +388,49 @@ describe("CloudManagedEndpointRuntime", () => { "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([600]); + 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("reports a connector that exits before Cloudflare registers it", () => + 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: 601, - all: Stream.never, - exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + pid, + ...(spawnCount === 1 + ? { + all: Stream.never, + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)), + } + : {}), onKill: () => { - killed.push(601); + killed.push(pid); }, }); yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + if (spawnCount === 2) { + yield* Deferred.succeed(secondSpawned, undefined); + } return handle; }), ); @@ -419,10 +448,56 @@ describe("CloudManagedEndpointRuntime", () => { 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 = []; diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index 4cf50b4ccb4d..14a177a3a285 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -134,7 +134,6 @@ export const make = Effect.gen(function* () { } satisfies CloudManagedEndpointRuntimeStatus; } - yield* stopActive; 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.`; @@ -256,73 +255,83 @@ export const make = Effect.gen(function* () { } 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), + 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 && 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 yield* awaitConnectorConnection(connector).pipe(Effect.onInterrupt(() => stopActive)); - } + 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 yield* restore(awaitConnectorConnection(connector)).pipe( + Effect.onInterrupt(() => stopActive), + ); + } - 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 { + 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")( From 808dc81410302bb63a67468fd5a545058636200e Mon Sep 17 00:00:00 2001 From: "Mateo M." Date: Thu, 27 Aug 2026 14:29:08 +0200 Subject: [PATCH 5/5] fix(server): release relay restart permit --- .../src/cloud/ManagedEndpointRuntime.test.ts | 62 ++++++ .../src/cloud/ManagedEndpointRuntime.ts | 203 +++++++++--------- 2 files changed, 169 insertions(+), 96 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 4de6c7cb53cb..9281d75d0bea 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -291,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 = []; diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index 14a177a3a285..73e9fbde095b 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -109,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* () { @@ -152,10 +156,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); @@ -178,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( @@ -222,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; @@ -240,97 +337,11 @@ export const make = Effect.gen(function* () { } 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; - } - - 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 yield* restore(awaitConnectorConnection(connector)).pipe( - Effect.onInterrupt(() => stopActive), - ); - } - - 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), ); });