authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 02:17:30-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 02:17:30-07:00
logcd45be3b57dc34796c21e7bbad9ae55fef5a583b
tree3436178c09b6560849522bd65bc92d76d5f45a18
parent9a5dc75648de725c737069379600570e391ec8b5
signature Signed by PGP key B5690EEEBB952194

fix(server): T3 Connect no longer stays down after a network change (#19335)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 203 insertions(+), 6 deletions(-)

apps/server/src/cloud/ManagedEndpointRuntime.test.ts+122-3
...@@ -3,6 +3,7 @@ import { vi } from "vite-plus/test";...@@ -3,6 +3,7 @@ import { vi } from "vite-plus/test";
3import * as Deferred from "effect/Deferred";3import * as Deferred from "effect/Deferred";
4import * as Duration from "effect/Duration";4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";7import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";8import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";9import * as Option from "effect/Option";
...@@ -12,6 +13,7 @@ import * as Sink from "effect/Sink";...@@ -12,6 +13,7 @@ import * as Sink from "effect/Sink";
12import * as Stream from "effect/Stream";13import * as Stream from "effect/Stream";
13import * as TestClock from "effect/testing/TestClock";14import * as TestClock from "effect/testing/TestClock";
14import { ChildProcess, ChildProcessSpawner } from "effect/process";15import { ChildProcess, ChildProcessSpawner } from "effect/process";
16import { HttpClient, HttpClientError, HttpClientResponse } from "effect/http";
15import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";17import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
16import * as RelayClient from "@t3tools/shared/relayClient";18import * as RelayClient from "@t3tools/shared/relayClient";
1719
...@@ -33,12 +35,17 @@ const layerRelayClientAvailable = Layer.succeed(...@@ -33,12 +35,17 @@ const layerRelayClientAvailable = Layer.succeed(
33 }),35 }),
34);36);
3537
38// Only connectors that log a metrics address are polled for readiness.
39const unusedHttpClient = HttpClient.make(() => Effect.die("unused"));
40
36const runtimeDependencies = (41const runtimeDependencies = (
37 spawner: ReturnType<typeof ChildProcessSpawner.make>,42 spawner: ReturnType<typeof ChildProcessSpawner.make>,
38 relayClientLayer = layerRelayClientAvailable,43 relayClientLayer = layerRelayClientAvailable,
44 httpClient = unusedHttpClient,
39) =>45) =>
40 Layer.mergeAll(46 Layer.mergeAll(
41 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),47 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),
48 Layer.succeed(HttpClient.HttpClient, httpClient),
42 relayClientLayer,49 relayClientLayer,
43 Layer.mock(ServerSecretStore.ServerSecretStore)({50 Layer.mock(ServerSecretStore.ServerSecretStore)({
44 get: () => Effect.succeedNone,51 get: () => Effect.succeedNone,
...@@ -48,11 +55,12 @@ const runtimeDependencies = (...@@ -48,11 +55,12 @@ const runtimeDependencies = (
48const buildCloudManagedEndpointRuntime = (55const buildCloudManagedEndpointRuntime = (
49 spawner: ReturnType<typeof ChildProcessSpawner.make>,56 spawner: ReturnType<typeof ChildProcessSpawner.make>,
50 relayClientLayer = layerRelayClientAvailable,57 relayClientLayer = layerRelayClientAvailable,
58 httpClient = unusedHttpClient,
51) =>59) =>
52 Effect.gen(function* () {60 Effect.gen(function* () {
53 const context = yield* Layer.build(61 const context = yield* Layer.build(
54 ManagedEndpointRuntime.layer.pipe(62 ManagedEndpointRuntime.layer.pipe(
55 Layer.provide(runtimeDependencies(spawner, relayClientLayer)),63 Layer.provide(runtimeDependencies(spawner, relayClientLayer, httpClient)),
56 ),64 ),
57 );65 );
58 return yield* Effect.service(ManagedEndpointRuntime.CloudManagedEndpointRuntime).pipe(66 return yield* Effect.service(ManagedEndpointRuntime.CloudManagedEndpointRuntime).pipe(
...@@ -410,6 +418,97 @@ describe("CloudManagedEndpointRuntime", () => {...@@ -410,6 +418,97 @@ describe("CloudManagedEndpointRuntime", () => {
410 }).pipe(Effect.provide(TestClock.layer())),418 }).pipe(Effect.provide(TestClock.layer())),
411 );419 );
412420
421 it.effect(
422 "restarts a running connector whose readiness check keeps reporting no connections",
423 () =>
424 Effect.gen(function* () {
425 const spawned: Array<number> = [];
426 const killed: Array<number> = [];
427 const exits: Array<Deferred.Deferred<ChildProcessSpawner.ExitCode>> = [];
428 const secondSpawned = yield* Deferred.make<void>();
429 const spawner = ChildProcessSpawner.make(() =>
430 Effect.gen(function* () {
431 const index = spawned.length;
432 const pid = 1000 + index;
433 const exit = yield* Deferred.make<ChildProcessSpawner.ExitCode>();
434 spawned.push(pid);
435 exits.push(exit);
436 if (index === 1) yield* Deferred.succeed(secondSpawned, undefined);
437 const handle = makeHandle({
438 pid,
439 exitCode: Deferred.await(exit),
440 onKill: () => {
441 if (!killed.includes(pid)) killed.push(pid);
442 Deferred.doneUnsafe(exit, Exit.succeed(ChildProcessSpawner.ExitCode(143)));
443 },
444 output: Stream.concat(
445 Stream.make(
446 new TextEncoder().encode(
447 `2026-10-14T08:49:04Z INF Starting metrics server on 127.0.0.1:${40000 + index}/metrics\n`,
448 ),
449 ),
450 Stream.never,
451 ),
452 });
453 yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
454 return handle;
455 }),
456 );
457 // Each readiness check takes the next scripted answer, then 503. It
458 // records which connectors were killed before it was sent.
459 const answers: Array<number | "refused"> = [];
460 const checks = yield* Queue.unbounded<{ readonly url: string; readonly killed: string }>();
461 const httpClient = HttpClient.make((request) => {
462 Queue.offerUnsafe(checks, { url: request.url, killed: killed.join(",") });
463 const status = answers.shift() ?? 503;
464 return status === "refused"
465 ? Effect.fail(
466 new HttpClientError.HttpClientError({
467 reason: new HttpClientError.TransportError({ request }),
468 }),
469 )
470 : Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status })));
471 });
472 // Advances the test clock until the watchdog checks readiness again.
473 const nextCheck = Effect.gen(function* () {
474 for (let second = 0; second < 60; second += 1) {
475 yield* TestClock.adjust(Duration.seconds(1));
476 const check = yield* Queue.poll(checks);
477 if (Option.isSome(check)) return check.value;
478 }
479 return yield* Effect.die("No readiness check within a minute");
480 });
481 const runtime = yield* buildCloudManagedEndpointRuntime(
482 spawner,
483 layerRelayClientAvailable,
484 httpClient,
485 );
486 yield* runtime.applyConfig({
487 providerKind: "cloudflare_tunnel",
488 connectorToken: "token",
489 tunnelId: "tunnel-1",
490 });
491
492 // Five 503s, a failed request that does not count, a ready answer that
493 // clears the streak, then six 503s in a row.
494 answers.push(503, 503, 503, 503, 503, "refused", 200);
495 for (let check = 0; check < 13; check += 1) {
496 expect(yield* nextCheck).toEqual({ url: "http://127.0.0.1:40000/ready", killed: "" });
497 }
498 yield* Deferred.await(secondSpawned);
499 expect(killed).toEqual([1000]);
500 expect(spawned).toEqual([1000, 1001]);
501
502 // The replacement follows a restart that never got a ready answer, so
503 // it is killed only after twelve 503s.
504 for (let check = 0; check < 12; check += 1) {
505 expect(yield* nextCheck).toEqual({ url: "http://127.0.0.1:40001/ready", killed: "1000" });
506 }
507 yield* Deferred.await(exits[1]!);
508 expect(killed).toEqual([1000, 1001]);
509 }).pipe(Effect.provide(TestClock.layer())),
510 );
511
413 it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () =>512 it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () =>
414 Effect.gen(function* () {513 Effect.gen(function* () {
415 const spawned: Array<ChildProcess.StandardCommand> = [];514 const spawned: Array<ChildProcess.StandardCommand> = [];
...@@ -457,8 +556,28 @@ describe("CloudManagedEndpointRuntime", () => {...@@ -457,8 +556,28 @@ describe("CloudManagedEndpointRuntime", () => {
457556
458 expect(spawned.map((command) => command.command)).toEqual(["cloudflared", "cloudflared"]);557 expect(spawned.map((command) => command.command)).toEqual(["cloudflared", "cloudflared"]);
459 expect(spawned.map((command) => command.args)).toEqual([558 expect(spawned.map((command) => command.args)).toEqual([
460 ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"],559 [
461 ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"],560 "tunnel",
561 "--no-autoupdate",
562 "--loglevel",
563 "info",
564 "--output",
565 "default",
566 "--metrics",
567 "127.0.0.1:0",
568 "run",
569 ],
570 [
571 "tunnel",
572 "--no-autoupdate",
573 "--loglevel",
574 "info",
575 "--output",
576 "default",
577 "--metrics",
578 "127.0.0.1:0",
579 "run",
580 ],
462 ]);581 ]);
463 expect(spawned.map((command) => command.options.env?.TUNNEL_TOKEN)).toEqual([582 expect(spawned.map((command) => command.options.env?.TUNNEL_TOKEN)).toEqual([
464 "token-1",583 "token-1",
apps/server/src/cloud/ManagedEndpointRuntime.ts+81-3
...@@ -7,12 +7,14 @@ import * as Duration from "effect/Duration";...@@ -7,12 +7,14 @@ import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";8import * as Exit from "effect/Exit";
9import * as Layer from "effect/Layer";9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
10import * as Queue from "effect/Queue";11import * as Queue from "effect/Queue";
11import * as Ref from "effect/Ref";12import * as Ref from "effect/Ref";
12import * as Result from "effect/Result";13import * as Result from "effect/Result";
13import * as Semaphore from "effect/Semaphore";14import * as Semaphore from "effect/Semaphore";
14import * as Scope from "effect/Scope";15import * as Scope from "effect/Scope";
15import * as Stream from "effect/Stream";16import * as Stream from "effect/Stream";
17import { HttpClient } from "effect/http";
16import * as ChildProcess from "effect/process/ChildProcess";18import * as ChildProcess from "effect/process/ChildProcess";
17import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";19import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
1820
...@@ -63,6 +65,8 @@ interface ActiveConnector {...@@ -63,6 +65,8 @@ interface ActiveConnector {
63 readonly startedAtMillis: number;65 readonly startedAtMillis: number;
64 /** Completes when the connector first registers a tunnel connection. */66 /** Completes when the connector first registers a tunnel connection. */
65 readonly registered: Deferred.Deferred<void>;67 readonly registered: Deferred.Deferred<void>;
68 /** Completes with the loopback address of the connector's metrics server. */
69 readonly metricsAddress: Deferred.Deferred<string>;
66}70}
6771
68// A connector that exits before running this long is treated as part of a72// 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;...@@ -80,6 +84,18 @@ const TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY = 4;
80// Ask for recovery after this long, and again at this interval while it stays84// Ask for recovery after this long, and again at this interval while it stays
81// unconnected; the relay hands back the same tunnel when it is still live.85// unconnected; the relay hands back the same tunnel when it is still live.
82const CONNECTOR_REGISTRATION_TIMEOUT = Duration.minutes(3);86const CONNECTOR_REGISTRATION_TIMEOUT = Duration.minutes(3);
87// cloudflared resolves the edge only at startup, so after a network change it
88// can stay alive with zero connections, dialing addresses that no longer route.
89// Its /ready endpoint answers 503 then. After this many 503s in a row (about
90// three minutes), the connector is restarted. Each restart that is never
91// followed by a ready answer doubles the count, up to about thirty minutes, so
92// a host that is really offline does not restart every few minutes.
93const CONNECTOR_READINESS_INTERVAL = Duration.seconds(30);
94const CONNECTOR_READINESS_REQUEST_TIMEOUT = Duration.seconds(5);
95const CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART = 6;
96const CONNECTOR_NOT_READY_CHECKS_MAX = 60;
97// The connector binds its metrics server to a random loopback port and logs it.
98const RELAY_CLIENT_METRICS_LINE = /\bStarting metrics server on (127\.0\.0\.1:\d+)\/metrics\b/u;
83// A failed background install of the pinned relay client is retried no sooner99// A failed background install of the pinned relay client is retried no sooner
84// than this, so a crash-looping connector or an offline host does not redownload100// than this, so a crash-looping connector or an offline host does not redownload
85// on every reconcile.101// on every reconcile.
...@@ -162,6 +178,7 @@ const stopConnector = (connector: ActiveConnector | null) =>...@@ -162,6 +178,7 @@ const stopConnector = (connector: ActiveConnector | null) =>
162export const make = Effect.gen(function* () {178export const make = Effect.gen(function* () {
163 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;179 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
164 const relayClient = yield* RelayClient.RelayClient;180 const relayClient = yield* RelayClient.RelayClient;
181 const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
165 const activeRef = yield* Ref.make<ActiveConnector | null>(null);182 const activeRef = yield* Ref.make<ActiveConnector | null>(null);
166 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);183 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
167 const recoveryRequests = yield* Queue.sliding<RelayManagedEndpointRuntimeConfig>(1);184 const recoveryRequests = yield* Queue.sliding<RelayManagedEndpointRuntimeConfig>(1);
...@@ -174,6 +191,7 @@ export const make = Effect.gen(function* () {...@@ -174,6 +191,7 @@ export const make = Effect.gen(function* () {
174 const lastInstallFailureAtRef = yield* Ref.make<number | null>(null);191 const lastInstallFailureAtRef = yield* Ref.make<number | null>(null);
175 const prunedRef = yield* Ref.make(false);192 const prunedRef = yield* Ref.make(false);
176 const retryLoopRunningRef = yield* Ref.make(false);193 const retryLoopRunningRef = yield* Ref.make(false);
194 const notReadyChecksBeforeRestartRef = yield* Ref.make(CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
177 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];195 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
178196
179 const stopActive = Effect.gen(function* () {197 const stopActive = Effect.gen(function* () {
...@@ -393,6 +411,48 @@ export const make = Effect.gen(function* () {...@@ -393,6 +411,48 @@ export const make = Effect.gen(function* () {
393 return false;411 return false;
394 }).pipe(Effect.repeat({ until: (connected) => connected }));412 }).pipe(Effect.repeat({ until: (connected) => connected }));
395413
414 // Kills a connector whose /ready keeps answering 503, so the supervisor
415 // restarts it like any exited connector. Only an explicit 503 counts: a failed
416 // request says nothing about the tunnel. A connector that never logs its
417 // metrics address is not polled.
418 const watchConnectorReadiness = (connector: ActiveConnector) =>
419 Effect.gen(function* () {
420 const readyUrl = `http://${yield* Deferred.await(connector.metricsAddress)}/ready`;
421 let notReadyChecks = 0;
422 while (true) {
423 yield* Effect.sleep(CONNECTOR_READINESS_INTERVAL);
424 const status = yield* httpClient.get(readyUrl).pipe(
425 Effect.map((response) => response.status),
426 Effect.scoped,
427 Effect.withTracerEnabled(false),
428 Effect.timeoutOption(CONNECTOR_READINESS_REQUEST_TIMEOUT),
429 Effect.orElseSucceed(() => Option.none<number>()),
430 );
431 if (Option.isNone(status)) continue;
432 if (status.value === 200) {
433 notReadyChecks = 0;
434 yield* Ref.set(notReadyChecksBeforeRestartRef, CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
435 continue;
436 }
437 if (status.value !== 503) continue;
438 notReadyChecks += 1;
439 const threshold = yield* Ref.get(notReadyChecksBeforeRestartRef);
440 if (notReadyChecks < threshold) continue;
441 notReadyChecks = 0;
442 yield* Ref.set(
443 notReadyChecksBeforeRestartRef,
444 Math.min(threshold * 2, CONNECTOR_NOT_READY_CHECKS_MAX),
445 );
446 yield* Effect.logWarning("Relay client has no tunnel connection; restarting it", {
447 pid: Number(connector.child.pid),
448 notReadyChecks: threshold,
449 tunnelId: connector.config.tunnelId,
450 tunnelName: connector.config.tunnelName,
451 });
452 yield* connector.child.kill().pipe(Effect.ignore);
453 }
454 });
455
396 const observeConnectorOutput = (connector: ActiveConnector) => {456 const observeConnectorOutput = (connector: ActiveConnector) => {
397 let rejectedRegistrations = 0;457 let rejectedRegistrations = 0;
398458
...@@ -443,8 +503,14 @@ export const make = Effect.gen(function* () {...@@ -443,8 +503,14 @@ export const make = Effect.gen(function* () {
443 }503 }
444 }504 }
445 return Effect.logWarning("Relay client reported a transport warning", attributes);505 return Effect.logWarning("Relay client reported a transport warning", attributes);
446 case "debug":506 case "debug": {
447 return Effect.logDebug("Relay client output", attributes);507 const metricsAddress = RELAY_CLIENT_METRICS_LINE.exec(line)?.[1];
508 return metricsAddress
509 ? Deferred.succeed(connector.metricsAddress, metricsAddress).pipe(
510 Effect.andThen(Effect.logDebug("Relay client output", attributes)),
511 )
512 : Effect.logDebug("Relay client output", attributes);
513 }
448 }514 }
449 }),515 }),
450 Effect.catchCause((cause) =>516 Effect.catchCause((cause) =>
...@@ -507,7 +573,17 @@ export const make = Effect.gen(function* () {...@@ -507,7 +573,17 @@ export const make = Effect.gen(function* () {
507 .spawn(573 .spawn(
508 ChildProcess.make(574 ChildProcess.make(
509 executable.executablePath,575 executable.executablePath,
510 ["tunnel", "--no-autoupdate", "--loglevel", "info", "--output", "default", "run"],576 [
577 "tunnel",
578 "--no-autoupdate",
579 "--loglevel",
580 "info",
581 "--output",
582 "default",
583 "--metrics",
584 "127.0.0.1:0",
585 "run",
586 ],
511 {587 {
512 detached: false,588 detached: false,
513 env: {589 env: {
...@@ -563,11 +639,13 @@ export const make = Effect.gen(function* () {...@@ -563,11 +639,13 @@ export const make = Effect.gen(function* () {
563 config,639 config,
564 startedAtMillis: yield* Clock.currentTimeMillis,640 startedAtMillis: yield* Clock.currentTimeMillis,
565 registered: yield* Deferred.make<void>(),641 registered: yield* Deferred.make<void>(),
642 metricsAddress: yield* Deferred.make<string>(),
566 } satisfies ActiveConnector;643 } satisfies ActiveConnector;
567 yield* Ref.set(activeRef, connector);644 yield* Ref.set(activeRef, connector);
568 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);645 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
569 yield* Effect.forkIn(superviseConnector(connector), connectorScope);646 yield* Effect.forkIn(superviseConnector(connector), connectorScope);
570 yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope);647 yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope);
648 yield* Effect.forkIn(watchConnectorReadiness(connector), connectorScope);
571 return {649 return {
572 status: "running",650 status: "running",
573 providerKind: "cloudflare_tunnel",651 providerKind: "cloudflare_tunnel",