From cd45be3b57dc34796c21e7bbad9ae55fef5a583b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 02:17:30 -0700 Subject: [PATCH] fix(server): T3 Connect no longer stays down after a network change (#19335) Co-authored-by: Claude Opus 5.5 --- .../src/cloud/ManagedEndpointRuntime.test.ts | 125 +++++++++++++++++- .../src/cloud/ManagedEndpointRuntime.ts | 84 +++++++++++- 2 files changed, 203 insertions(+), 6 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 19ada42646835cf62b91a085f82b5620c5d187d9..b4a6122a443e3bfcb3749b2b2b10d22925b7d8d5 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -3,6 +3,7 @@ import { vi } from "vite-plus/test"; 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"; @@ -12,6 +13,7 @@ import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { HttpClient, HttpClientError, HttpClientResponse } from "effect/http"; import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -33,12 +35,17 @@ const layerRelayClientAvailable = Layer.succeed( }), ); +// Only connectors that log a metrics address are polled for readiness. +const unusedHttpClient = HttpClient.make(() => Effect.die("unused")); + const runtimeDependencies = ( spawner: ReturnType, relayClientLayer = layerRelayClientAvailable, + httpClient = unusedHttpClient, ) => Layer.mergeAll( Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed(HttpClient.HttpClient, httpClient), relayClientLayer, Layer.mock(ServerSecretStore.ServerSecretStore)({ get: () => Effect.succeedNone, @@ -48,11 +55,12 @@ const runtimeDependencies = ( const buildCloudManagedEndpointRuntime = ( spawner: ReturnType, relayClientLayer = layerRelayClientAvailable, + httpClient = unusedHttpClient, ) => Effect.gen(function* () { const context = yield* Layer.build( ManagedEndpointRuntime.layer.pipe( - Layer.provide(runtimeDependencies(spawner, relayClientLayer)), + Layer.provide(runtimeDependencies(spawner, relayClientLayer, httpClient)), ), ); return yield* Effect.service(ManagedEndpointRuntime.CloudManagedEndpointRuntime).pipe( @@ -410,6 +418,97 @@ describe("CloudManagedEndpointRuntime", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect( + "restarts a running connector whose readiness check keeps reporting no connections", + () => + Effect.gen(function* () { + const spawned: Array = []; + const killed: Array = []; + const exits: Array> = []; + const secondSpawned = yield* Deferred.make(); + const spawner = ChildProcessSpawner.make(() => + Effect.gen(function* () { + const index = spawned.length; + const pid = 1000 + index; + const exit = yield* Deferred.make(); + spawned.push(pid); + exits.push(exit); + if (index === 1) yield* Deferred.succeed(secondSpawned, undefined); + const handle = makeHandle({ + pid, + exitCode: Deferred.await(exit), + onKill: () => { + if (!killed.includes(pid)) killed.push(pid); + Deferred.doneUnsafe(exit, Exit.succeed(ChildProcessSpawner.ExitCode(143))); + }, + output: Stream.concat( + Stream.make( + new TextEncoder().encode( + `2026-10-14T08:49:04Z INF Starting metrics server on 127.0.0.1:${40000 + index}/metrics\n`, + ), + ), + Stream.never, + ), + }); + yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); + return handle; + }), + ); + // Each readiness check takes the next scripted answer, then 503. It + // records which connectors were killed before it was sent. + const answers: Array = []; + const checks = yield* Queue.unbounded<{ readonly url: string; readonly killed: string }>(); + const httpClient = HttpClient.make((request) => { + Queue.offerUnsafe(checks, { url: request.url, killed: killed.join(",") }); + const status = answers.shift() ?? 503; + return status === "refused" + ? Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ request }), + }), + ) + : Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status }))); + }); + // Advances the test clock until the watchdog checks readiness again. + const nextCheck = Effect.gen(function* () { + for (let second = 0; second < 60; second += 1) { + yield* TestClock.adjust(Duration.seconds(1)); + const check = yield* Queue.poll(checks); + if (Option.isSome(check)) return check.value; + } + return yield* Effect.die("No readiness check within a minute"); + }); + const runtime = yield* buildCloudManagedEndpointRuntime( + spawner, + layerRelayClientAvailable, + httpClient, + ); + yield* runtime.applyConfig({ + providerKind: "cloudflare_tunnel", + connectorToken: "token", + tunnelId: "tunnel-1", + }); + + // Five 503s, a failed request that does not count, a ready answer that + // clears the streak, then six 503s in a row. + answers.push(503, 503, 503, 503, 503, "refused", 200); + for (let check = 0; check < 13; check += 1) { + expect(yield* nextCheck).toEqual({ url: "http://127.0.0.1:40000/ready", killed: "" }); + } + yield* Deferred.await(secondSpawned); + expect(killed).toEqual([1000]); + expect(spawned).toEqual([1000, 1001]); + + // The replacement follows a restart that never got a ready answer, so + // it is killed only after twelve 503s. + for (let check = 0; check < 12; check += 1) { + expect(yield* nextCheck).toEqual({ url: "http://127.0.0.1:40001/ready", killed: "1000" }); + } + yield* Deferred.await(exits[1]!); + expect(killed).toEqual([1000, 1001]); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () => Effect.gen(function* () { const spawned: Array = []; @@ -457,8 +556,28 @@ describe("CloudManagedEndpointRuntime", () => { expect(spawned.map((command) => command.command)).toEqual(["cloudflared", "cloudflared"]); expect(spawned.map((command) => command.args)).toEqual([ - ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"], - ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"], + [ + "tunnel", + "--no-autoupdate", + "--loglevel", + "info", + "--output", + "default", + "--metrics", + "127.0.0.1:0", + "run", + ], + [ + "tunnel", + "--no-autoupdate", + "--loglevel", + "info", + "--output", + "default", + "--metrics", + "127.0.0.1:0", + "run", + ], ]); expect(spawned.map((command) => command.options.env?.TUNNEL_TOKEN)).toEqual([ "token-1", diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index a774c77550838b8aef1bdeab8e2acae0861ede94..0365cc28e6194512e7174fd651bc1fabd2809773 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -7,12 +7,14 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import { HttpClient } from "effect/http"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; @@ -63,6 +65,8 @@ interface ActiveConnector { readonly startedAtMillis: number; /** Completes when the connector first registers a tunnel connection. */ readonly registered: Deferred.Deferred; + /** Completes with the loopback address of the connector's metrics server. */ + readonly metricsAddress: Deferred.Deferred; } // A connector that exits before running this long is treated as part of a @@ -80,6 +84,18 @@ const TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY = 4; // Ask for recovery after this long, and again at this interval while it stays // unconnected; the relay hands back the same tunnel when it is still live. const CONNECTOR_REGISTRATION_TIMEOUT = Duration.minutes(3); +// cloudflared resolves the edge only at startup, so after a network change it +// can stay alive with zero connections, dialing addresses that no longer route. +// Its /ready endpoint answers 503 then. After this many 503s in a row (about +// three minutes), the connector is restarted. Each restart that is never +// followed by a ready answer doubles the count, up to about thirty minutes, so +// a host that is really offline does not restart every few minutes. +const CONNECTOR_READINESS_INTERVAL = Duration.seconds(30); +const CONNECTOR_READINESS_REQUEST_TIMEOUT = Duration.seconds(5); +const CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART = 6; +const CONNECTOR_NOT_READY_CHECKS_MAX = 60; +// The connector binds its metrics server to a random loopback port and logs it. +const RELAY_CLIENT_METRICS_LINE = /\bStarting metrics server on (127\.0\.0\.1:\d+)\/metrics\b/u; // A failed background install of the pinned relay client is retried no sooner // than this, so a crash-looping connector or an offline host does not redownload // on every reconcile. @@ -162,6 +178,7 @@ const stopConnector = (connector: ActiveConnector | null) => export const make = Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const relayClient = yield* RelayClient.RelayClient; + const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient); const activeRef = yield* Ref.make(null); const desiredConfigRef = yield* Ref.make(null); const recoveryRequests = yield* Queue.sliding(1); @@ -174,6 +191,7 @@ export const make = Effect.gen(function* () { const lastInstallFailureAtRef = yield* Ref.make(null); const prunedRef = yield* Ref.make(false); const retryLoopRunningRef = yield* Ref.make(false); + const notReadyChecksBeforeRestartRef = yield* Ref.make(CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART); let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"]; const stopActive = Effect.gen(function* () { @@ -393,6 +411,48 @@ export const make = Effect.gen(function* () { return false; }).pipe(Effect.repeat({ until: (connected) => connected })); + // Kills a connector whose /ready keeps answering 503, so the supervisor + // restarts it like any exited connector. Only an explicit 503 counts: a failed + // request says nothing about the tunnel. A connector that never logs its + // metrics address is not polled. + const watchConnectorReadiness = (connector: ActiveConnector) => + Effect.gen(function* () { + const readyUrl = `http://${yield* Deferred.await(connector.metricsAddress)}/ready`; + let notReadyChecks = 0; + while (true) { + yield* Effect.sleep(CONNECTOR_READINESS_INTERVAL); + const status = yield* httpClient.get(readyUrl).pipe( + Effect.map((response) => response.status), + Effect.scoped, + Effect.withTracerEnabled(false), + Effect.timeoutOption(CONNECTOR_READINESS_REQUEST_TIMEOUT), + Effect.orElseSucceed(() => Option.none()), + ); + if (Option.isNone(status)) continue; + if (status.value === 200) { + notReadyChecks = 0; + yield* Ref.set(notReadyChecksBeforeRestartRef, CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART); + continue; + } + if (status.value !== 503) continue; + notReadyChecks += 1; + const threshold = yield* Ref.get(notReadyChecksBeforeRestartRef); + if (notReadyChecks < threshold) continue; + notReadyChecks = 0; + yield* Ref.set( + notReadyChecksBeforeRestartRef, + Math.min(threshold * 2, CONNECTOR_NOT_READY_CHECKS_MAX), + ); + yield* Effect.logWarning("Relay client has no tunnel connection; restarting it", { + pid: Number(connector.child.pid), + notReadyChecks: threshold, + tunnelId: connector.config.tunnelId, + tunnelName: connector.config.tunnelName, + }); + yield* connector.child.kill().pipe(Effect.ignore); + } + }); + const observeConnectorOutput = (connector: ActiveConnector) => { let rejectedRegistrations = 0; @@ -443,8 +503,14 @@ export const make = Effect.gen(function* () { } } return Effect.logWarning("Relay client reported a transport warning", attributes); - case "debug": - return Effect.logDebug("Relay client output", attributes); + case "debug": { + const metricsAddress = RELAY_CLIENT_METRICS_LINE.exec(line)?.[1]; + return metricsAddress + ? Deferred.succeed(connector.metricsAddress, metricsAddress).pipe( + Effect.andThen(Effect.logDebug("Relay client output", attributes)), + ) + : Effect.logDebug("Relay client output", attributes); + } } }), Effect.catchCause((cause) => @@ -507,7 +573,17 @@ export const make = Effect.gen(function* () { .spawn( ChildProcess.make( executable.executablePath, - ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"], + [ + "tunnel", + "--no-autoupdate", + "--loglevel", + "info", + "--output", + "default", + "--metrics", + "127.0.0.1:0", + "run", + ], { detached: false, env: { @@ -563,11 +639,13 @@ export const make = Effect.gen(function* () { config, startedAtMillis: yield* Clock.currentTimeMillis, registered: yield* Deferred.make(), + metricsAddress: yield* Deferred.make(), } satisfies ActiveConnector; yield* Ref.set(activeRef, connector); yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); yield* Effect.forkIn(superviseConnector(connector), connectorScope); yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope); + yield* Effect.forkIn(watchConnectorReadiness(connector), connectorScope); return { status: "running", providerKind: "cloudflare_tunnel", -- 2.54.0