Open History to see the last commit
1import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
2import { RotatingFileSink, RotatingFileSinkError } from "@t3tools/shared/logging";
3import * as HostProcess from "@t3tools/shared/HostProcess";
4import * as RelayClient from "@t3tools/shared/relayClient";
5import * as Clock from "effect/Clock";
6import * as Config from "effect/Config";
7import * as Context from "effect/Context";
8import * as Deferred from "effect/Deferred";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Exit from "effect/Exit";
12import * as Layer from "effect/Layer";
13import * as Logger from "effect/Logger";
14import * as Option from "effect/Option";
15import * as Path from "effect/Path";
16import * as Queue from "effect/Queue";
17import * as Ref from "effect/Ref";
18import * as Result from "effect/Result";
19import * as Schema from "effect/Schema";
20import * as Semaphore from "effect/Semaphore";
21import * as Scope from "effect/Scope";
22import * as Stream from "effect/Stream";
23import { HttpClient, HttpClientResponse } from "effect/http";
24import * as ChildProcess from "effect/process/ChildProcess";
25import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
26
27import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
28import * as ServerConfig from "../config.ts";
29import * as ManagedLimits from "../policy/managedLimits.ts";
30import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
31
32export type CloudManagedEndpointRuntimeStatus =
33 | {
34 readonly status: "disabled";
35 }
36 | {
37 readonly status: "failed";
38 readonly providerKind: RelayManagedEndpointRuntimeConfig["providerKind"];
39 readonly failure: "unsupported-platform" | "not-installed" | "spawn-failed";
40 readonly reason: string;
41 readonly tunnelId?: string;
42 readonly tunnelName?: string;
43 }
44 | {
45 readonly status: "running";
46 readonly providerKind: "cloudflare_tunnel";
47 readonly pid: number;
48 readonly tunnelId?: string;
49 readonly tunnelName?: string;
50 }
51 | {
52 readonly status: "unsupported";
53 readonly providerKind: RelayManagedEndpointRuntimeConfig["providerKind"];
54 };
55
56/** What the running cloudflared reports about itself. `null` fields are not known yet. */
57export interface ConnectorFacts {
58 readonly pid: number;
59 readonly readyConnections: number | null;
60 /** The public tunnel hostname from the remote config cloudflared receives once it connects. */
61 readonly hostname: string | null;
62 readonly protocol: string | null;
63}
64
65export class CloudManagedEndpointRuntime extends Context.Service<
66 CloudManagedEndpointRuntime,
67 {
68 readonly applyConfig: (
69 config: RelayManagedEndpointRuntimeConfig | null,
70 ) => Effect.Effect<CloudManagedEndpointRuntimeStatus>;
71 /** `null` when no connector is running. */
72 readonly inspectConnector: Effect.Effect<ConnectorFacts | null>;
73 /** Emits when a connector registers a connection or its readiness changes. */
74 readonly readinessChanged: Stream.Stream<void>;
75 readonly recoveryRequests: Stream.Stream<RelayManagedEndpointRuntimeConfig>;
76 readonly requestRecovery: (config: RelayManagedEndpointRuntimeConfig) => Effect.Effect<void>;
77 /**
78 * Restarts the connector if it still runs as `pid` and `stillDead` holds once it has
79 * the lock. Returns whether it sent the restart.
80 */
81 readonly restartConnector: (
82 pid: number,
83 stillDead: Effect.Effect<boolean>,
84 ) => Effect.Effect<boolean>;
85 readonly tunnelConnected: Stream.Stream<void>;
86 readonly withLinkStateLock: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
87 }
88>()("t3/cloud/ManagedEndpointRuntime/CloudManagedEndpointRuntime") {}
89
90interface ActiveConnector {
91 readonly child: ChildProcessSpawner.ChildProcessHandle;
92 readonly executable: RelayClient.AvailableRelayClient;
93 readonly scope: Scope.Closeable;
94 readonly configKey: string;
95 readonly config: RelayManagedEndpointRuntimeConfig;
96 readonly startedAtMillis: number;
97 readonly registered: Deferred.Deferred<void>;
98 readonly metricsAddress: Deferred.Deferred<string>;
99 readonly protocol: Ref.Ref<string | null>;
100}
101
102// An instantly failing relay client otherwise respawns ~100 times/sec and exhausts V8 via tracing.
103const isRotatingFileSinkError = Schema.is(RotatingFileSinkError);
104
105const RELAY_RESTART_STABLE_UPTIME_MS = 30_000;
106const RELAY_RESTART_BACKOFF_BASE_MS = 1_000;
107const RELAY_RESTART_BACKOFF_MAX_MS = 60_000;
108// Newly created tunnels can fail authorization briefly while Cloudflare propagates their token.
109const TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY = 4;
110// Unknown rejection wording can leave a dead tunnel unconnected; periodically request recovery.
111const CONNECTOR_REGISTRATION_TIMEOUT = Duration.minutes(3);
112// cloudflared resolves edges only at startup and can keep dialing stale addresses after a network change.
113const CONNECTOR_READINESS_INTERVAL = Duration.seconds(30);
114const CONNECTOR_READINESS_REQUEST_TIMEOUT = Duration.seconds(5);
115const CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART = 6;
116const CONNECTOR_NOT_READY_CHECKS_MAX = 60;
117const RELAY_CLIENT_METRICS_LINE = /\bStarting metrics server on (127\.0\.0\.1:\d+)\/metrics\b/u;
118const RELAY_CLIENT_PROTOCOL = /\bprotocol=(\w+)/u;
119// cloudflared's metrics server: `/ready` answers 200 or 503 with this body, and
120// `/config` holds the ingress rules the edge pushed after the first connection.
121const ConnectorReadyBody = Schema.Struct({ readyConnections: Schema.Number });
122const ConnectorConfigBody = Schema.Struct({
123 config: Schema.Struct({
124 ingress: Schema.optional(
125 Schema.NullOr(Schema.Array(Schema.Struct({ hostname: Schema.optional(Schema.String) }))),
126 ),
127 }),
128});
129// Registration resets cloudflared's own fallback, even when QUIC stalls again
130// immediately. Keep that evidence across registrations and connector restarts.
131const QUIC_STALLS_BEFORE_HTTP2 = 30;
132const QUIC_STALL_WINDOW_MS = 5 * 60_000;
133// Retry QUIC on a later startup after transient failures or network changes; reuse never renews this hint.
134const HTTP2_FALLBACK_TTL_MS = 24 * 60 * 60_000;
135const HTTP2_FALLBACK_STATE = "cloud-endpoint-http2-fallback";
136const Http2Fallback = Schema.fromJsonString(Schema.Struct({ fellBackAt: Schema.Number }));
137const decodeHttp2Fallback = Schema.decodeUnknownOption(Http2Fallback);
138const encodeHttp2Fallback = Schema.encodeSync(Http2Fallback);
139// A failed background install of the pinned relay client is retried no sooner
140// than this, so a crash-looping connector or an offline host does not redownload
141// on every reconcile.
142const RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS = 10 * 60_000;
143export const RELAY_CLIENT_LOG_FILE = "relay-client.ndjson";
144
145/** Preserve user-chosen PATH binaries and overrides; CLI downloads require consent. */
146function needsPinnedRelayClient(executable: RelayClient.RelayClientStatus): boolean {
147 return (
148 executable.status === "missing" ||
149 (executable.status === "available" &&
150 executable.source === "managed" &&
151 !RelayClient.isPinnedManagedRelayClient(executable))
152 );
153}
154
155export function classifyRelayClientOutput(
156 line: string,
157): "connected" | "unregistered" | "warning" | "debug" {
158 if (/\bRegistered tunnel connection\b/iu.test(line)) {
159 return "connected";
160 }
161 // cloudflared uses zerolog FTL/PNC for fatal/panic; classify them above debug too.
162 if (/\b(?:ERR|WRN|FTL|PNC)\b/u.test(line)) {
163 return "warning";
164 }
165 return /\bINF\b.*\bUnregistered tunnel connection\b/u.test(line) ? "unregistered" : "debug";
166}
167
168/** Edge rejection wording varies; any Unauthorized registration error counts. */
169export function isRejectedRelayClientTunnelOutput(line: string): boolean {
170 return (
171 /\bRegister tunnel error from server side\b/iu.test(line) &&
172 /error="(?:Unauthorized:[^"]*|Failed to get tunnel|Tunnel not found|Record for tunnel not found|Invalid tunnel secret)"/iu.test(
173 line,
174 )
175 );
176}
177
178export function isRetryableManagedEndpointRuntimeStatus(status: unknown): boolean {
179 if (typeof status !== "object" || status === null || !("status" in status)) {
180 return false;
181 }
182 if (status.status !== "failed" || !("failure" in status)) {
183 return false;
184 }
185 return status.failure === "not-installed" || status.failure === "spawn-failed";
186}
187
188function runtimeConfigKey(config: RelayManagedEndpointRuntimeConfig): string {
189 return JSON.stringify({
190 providerKind: config.providerKind,
191 connectorToken: config.connectorToken,
192 tunnelId: config.tunnelId ?? null,
193 tunnelName: config.tunnelName ?? null,
194 });
195}
196
197const stopConnector = (connector: ActiveConnector | null) =>
198 connector
199 ? Scope.close(connector.scope, Exit.void).pipe(
200 Effect.tap(() =>
201 Effect.logInfo("Relay client stopped", {
202 pid: Number(connector.child.pid),
203 }),
204 ),
205 Effect.ignore,
206 )
207 : Effect.void;
208
209/** @public */
210export const make = Effect.gen(function* () {
211 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
212 const relayClient = yield* RelayClient.RelayClient;
213 const secrets = yield* ServerSecretStore.ServerSecretStore;
214 const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
215 // Without T3 Connect no connector ever runs, whoever asks for one.
216 const connectAllowed = ManagedLimits.allowsConnect(yield* ManagedPolicy.ManagedPolicy);
217 const activeRef = yield* Ref.make<ActiveConnector | null>(null);
218 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
219 const recoveryRequests = yield* Queue.sliding<RelayManagedEndpointRuntimeConfig>(1);
220 const tunnelConnections = yield* Queue.sliding<void>(1);
221 const readinessChanges = yield* Queue.sliding<void>(1);
222 const reconcileSemaphore = yield* Semaphore.make(1);
223 const restartDelayRef = yield* Ref.make(0);
224 const linkStateSemaphore = yield* Semaphore.make(1);
225 const runtimeScope = yield* Effect.scope;
226 const installInFlightRef = yield* Ref.make(false);
227 const lastInstallFailureAtRef = yield* Ref.make<number | null>(null);
228 const prunedRef = yield* Ref.make(false);
229 const retryLoopRunningRef = yield* Ref.make(false);
230 const notReadyChecksBeforeRestartRef = yield* Ref.make(CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
231 // The relay client shares the server's process group, so it exits on these signals before the finalizer runs.
232 let shutdownSignalReceived = false;
233 const onShutdownSignal = () => {
234 shutdownSignalReceived = true;
235 };
236 yield* Effect.acquireRelease(
237 Effect.sync(() => {
238 process.on("SIGINT", onShutdownSignal);
239 process.on("SIGTERM", onShutdownSignal);
240 }),
241 () =>
242 Effect.sync(() => {
243 process.off("SIGINT", onShutdownSignal);
244 process.off("SIGTERM", onShutdownSignal);
245 }),
246 );
247 // An explicit cloudflared transport in the environment is the user's choice.
248 const env = yield* HostProcess.Environment;
249 const transportChosenByUser = Option.isSome(
250 yield* Config.String("TUNNEL_TRANSPORT_PROTOCOL").pipe(Config.option),
251 );
252 const previousFallback = transportChosenByUser
253 ? Option.none()
254 : yield* secrets.get(HTTP2_FALLBACK_STATE).pipe(
255 Effect.map(Option.flatMap((bytes) => decodeHttp2Fallback(new TextDecoder().decode(bytes)))),
256 Effect.orElseSucceed(() => Option.none()),
257 );
258 const now = yield* Clock.currentTimeMillis;
259 const http2Ref = yield* Ref.make(
260 Option.isSome(previousFallback) &&
261 now >= previousFallback.value.fellBackAt &&
262 now - previousFallback.value.fellBackAt < HTTP2_FALLBACK_TTL_MS,
263 );
264 const quicStallsRef = yield* Ref.make<ReadonlyArray<number>>([]);
265 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
266
267 const { logsDir } = yield* ServerConfig.ServerConfig;
268 const path = yield* Path.Path;
269 const relayClientLogPath = path.join(logsDir, RELAY_CLIENT_LOG_FILE);
270 // Synchronous appends avoid a flush timer for this low-volume log.
271 const fallbackLoggers = yield* Effect.service(Logger.CurrentLoggers);
272 let writeFailureReported = false;
273 const relayClientLog = yield* Effect.try({
274 try: () =>
275 new RotatingFileSink({
276 filePath: relayClientLogPath,
277 maxBytes: 1024 * 1024,
278 maxFiles: 2,
279 throwOnError: true,
280 }),
281 catch: (cause) =>
282 isRotatingFileSinkError(cause)
283 ? cause
284 : new RotatingFileSinkError({
285 operation: "initialize",
286 filePath: relayClientLogPath,
287 cause,
288 }),
289 }).pipe(
290 Effect.map((sink) =>
291 Logger.make((options) => {
292 try {
293 const entry = Logger.formatJson.log(options);
294 const bytes = Buffer.byteLength(entry) + 1;
295 sink.write(
296 bytes <= 1024 * 1024
297 ? `${entry}\n`
298 : `${JSON.stringify({
299 timestamp: options.date.toISOString(),
300 level: options.logLevel.toUpperCase(),
301 message: ["Relay client log entry omitted: exceeds 1 MiB", { bytes }],
302 })}\n`,
303 );
304 } catch {
305 if (!writeFailureReported) {
306 writeFailureReported = true;
307 for (const logger of fallbackLoggers) {
308 logger.log({
309 ...options,
310 logLevel: "Warn",
311 message: ["Relay client log unavailable", { filePath: relayClientLogPath }],
312 });
313 }
314 }
315 }
316 }),
317 ),
318 Effect.catch((error) =>
319 Effect.logWarning("Relay client log unavailable", {
320 filePath: relayClientLogPath,
321 cause: error,
322 }).pipe(Effect.as(undefined)),
323 ),
324 );
325 // Forked connector fibers inherit this logger after the applyConfig span ends.
326 const withRelayClientLog = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
327 relayClientLog ? Effect.withLogger(effect, relayClientLog) : effect;
328
329 const stopActive = Effect.gen(function* () {
330 const active = yield* Ref.getAndSet(activeRef, null);
331 yield* stopConnector(active);
332 });
333
334 // Lives until connector exit and starts replacements; avoid nested lifetime spans.
335 const superviseConnector = Effect.fnUntraced(
336 function* (connector: ActiveConnector) {
337 const result = yield* Effect.result(connector.child.exitCode);
338 const activeAtExit = yield* Ref.get(activeRef);
339 if (
340 activeAtExit?.child.pid !== connector.child.pid ||
341 activeAtExit.configKey !== connector.configKey
342 ) {
343 return;
344 }
345 const uptimeMillis = (yield* Clock.currentTimeMillis) - connector.startedAtMillis;
346 // Back off outside the semaphore so config changes remain responsive; recheck config afterwards.
347 const restartDelayMillis = yield* Ref.modify(restartDelayRef, (current) => {
348 if (uptimeMillis >= RELAY_RESTART_STABLE_UPTIME_MS) {
349 return [0, 0];
350 }
351 return [
352 current,
353 current === 0
354 ? RELAY_RESTART_BACKOFF_BASE_MS
355 : Math.min(current * 2, RELAY_RESTART_BACKOFF_MAX_MS),
356 ];
357 });
358 if (restartDelayMillis > 0) {
359 yield* Effect.logWarning("Relay client is crash-looping; delaying restart", {
360 pid: Number(connector.child.pid),
361 uptimeMillis,
362 restartDelayMillis,
363 tunnelId: connector.config.tunnelId,
364 tunnelName: connector.config.tunnelName,
365 });
366 yield* Effect.sleep(Duration.millis(restartDelayMillis));
367 }
368 yield* reconcileSemaphore.withPermits(1)(
369 Effect.gen(function* () {
370 const active = yield* Ref.get(activeRef);
371 if (
372 active?.child.pid !== connector.child.pid ||
373 active.configKey !== connector.configKey
374 ) {
375 return;
376 }
377 yield* Ref.set(activeRef, null);
378 yield* stopConnector(connector);
379
380 const desiredConfig = yield* Ref.get(desiredConfigRef);
381 if (
382 shutdownSignalReceived ||
383 !desiredConfig ||
384 desiredConfig.providerKind !== "cloudflare_tunnel" ||
385 runtimeConfigKey(desiredConfig) !== connector.configKey
386 ) {
387 return;
388 }
389
390 yield* Effect.logWarning("Relay client exited; restarting", {
391 pid: Number(connector.child.pid),
392 ...(Result.isSuccess(result)
393 ? { exitCode: Number(result.success) }
394 : { cause: result.failure }),
395 tunnelId: connector.config.tunnelId,
396 tunnelName: connector.config.tunnelName,
397 });
398 yield* Queue.offer(recoveryRequests, connector.config);
399 yield* reconcileConfig(desiredConfig);
400 }),
401 );
402 },
403 Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })),
404 );
405
406 // Keep the old connector until its replacement registers, preserving the relay connection.
407 const installPinnedRelayClient = Effect.gen(function* () {
408 if (yield* Ref.getAndSet(installInFlightRef, true)) return;
409 yield* Effect.logInfo("Installing the pinned relay client", {
410 version: RelayClient.CLOUDFLARED_VERSION,
411 });
412 yield* relayClient.install.pipe(
413 Effect.tap(() => Ref.set(lastInstallFailureAtRef, null)),
414 Effect.flatMap((installed) =>
415 reconcileSemaphore.withPermits(1)(
416 Effect.gen(function* () {
417 const desiredConfig = yield* Ref.get(desiredConfigRef);
418 if (!desiredConfig || desiredConfig.providerKind !== "cloudflare_tunnel") return;
419 const active = yield* Ref.get(activeRef);
420 // Self-updated binaries share the pinned path; compare versions to detect replacements.
421 if (
422 active?.executable.executablePath === installed.executablePath &&
423 active.executable.version === installed.version
424 ) {
425 return;
426 }
427 // A scanner can block a fresh binary's probe; retry instead of restarting onto the older binary.
428 const resolved = yield* relayClient.resolve;
429 if (
430 resolved.status !== "available" ||
431 resolved.executablePath !== installed.executablePath ||
432 resolved.version !== installed.version
433 ) {
434 return;
435 }
436 yield* Effect.logInfo("Relay client installed; moving the connector to it", {
437 version: installed.version,
438 previousVersion: active?.executable.version,
439 });
440 const previous = yield* Ref.getAndSet(activeRef, null);
441 const status = yield* reconcileConfig(desiredConfig);
442 const next = yield* Ref.get(activeRef);
443 if (!next) {
444 yield* Effect.logWarning("Relay client did not start after the update", status);
445 // The supervisor skips exits while detached; request recovery explicitly.
446 const previousRunning = previous
447 ? yield* previous.child.isRunning.pipe(Effect.orElseSucceed(() => false))
448 : false;
449 if (previous && previousRunning) {
450 yield* Ref.set(activeRef, previous);
451 } else {
452 yield* stopConnector(previous);
453 yield* Queue.offer(recoveryRequests, desiredConfig);
454 }
455 return;
456 }
457 if (previous) {
458 yield* Deferred.await(next.registered).pipe(
459 Effect.ensuring(stopConnector(previous)),
460 Effect.forkIn(next.scope),
461 );
462 }
463 }),
464 ),
465 ),
466 Effect.catchTags({
467 RelayClientInstallError: (error) =>
468 Clock.currentTimeMillis.pipe(
469 Effect.flatMap((failedAt) => Ref.set(lastInstallFailureAtRef, failedAt)),
470 Effect.andThen(
471 Effect.logWarning("Could not install the pinned relay client", {
472 reason: error.reason,
473 message: error.message,
474 cause: error.cause,
475 }),
476 ),
477 ),
478 }),
479 Effect.ensuring(Ref.set(installInFlightRef, false)),
480 );
481 });
482
483 const ensurePinnedRelayClient = Effect.gen(function* () {
484 const now = yield* Clock.currentTimeMillis;
485 const lastFailureAt = yield* Ref.get(lastInstallFailureAtRef);
486 if (lastFailureAt !== null && now - lastFailureAt < RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS) {
487 return;
488 }
489 yield* installPinnedRelayClient;
490 }).pipe(Effect.forkIn(runtimeScope), Effect.asVoid);
491
492 // Retry installs skipped by applyConfig without the failure gate.
493 const retryPinnedRelayClientInstall = Effect.gen(function* () {
494 while (true) {
495 yield* Effect.sleep(Duration.millis(RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS));
496 const desiredConfig = yield* Ref.get(desiredConfigRef);
497 if (!desiredConfig || desiredConfig.providerKind !== "cloudflare_tunnel") return;
498 // An installed pin still needs activation after a failed probe or while an older connector runs.
499 const active = yield* Ref.get(activeRef);
500 if (active && !needsPinnedRelayClient(active.executable)) return;
501 yield* installPinnedRelayClient;
502 }
503 }).pipe(Effect.ensuring(Ref.set(retryLoopRunningRef, false)));
504
505 const startPinnedRelayClientRetries = Effect.gen(function* () {
506 if (yield* Ref.getAndSet(retryLoopRunningRef, true)) return;
507 yield* Effect.forkIn(retryPinnedRelayClientInstall, runtimeScope);
508 });
509
510 // Registration retries can last for the connector lifetime; keep this loop untraced.
511 const watchConnectorRegistration = Effect.fnUntraced(
512 function* (connector: ActiveConnector) {
513 yield* Effect.sleep(CONNECTOR_REGISTRATION_TIMEOUT);
514 if (yield* Deferred.isDone(connector.registered)) return true;
515 yield* Effect.logWarning(
516 "Relay client has not registered a tunnel connection; requesting recovery",
517 {
518 pid: Number(connector.child.pid),
519 tunnelId: connector.config.tunnelId,
520 tunnelName: connector.config.tunnelName,
521 },
522 );
523 yield* Queue.offer(recoveryRequests, connector.config);
524 return false;
525 },
526 Effect.repeat({ until: (connected) => connected }),
527 );
528
529 // Only an explicit 503 counts: a failed request says nothing about the
530 // Readiness polling runs for the connector lifetime; keep this loop untraced.
531 const watchConnectorReadiness = Effect.fnUntraced(function* (connector: ActiveConnector) {
532 const readyUrl = `http://${yield* Deferred.await(connector.metricsAddress)}/ready`;
533 let notReadyChecks = 0;
534 let lastReady: boolean | null = null;
535 while (true) {
536 yield* Effect.sleep(CONNECTOR_READINESS_INTERVAL);
537 const status = yield* httpClient.get(readyUrl).pipe(
538 Effect.map((response) => response.status),
539 Effect.scoped,
540 Effect.withTracerEnabled(false),
541 Effect.timeoutOption(CONNECTOR_READINESS_REQUEST_TIMEOUT),
542 Effect.orElseSucceed(() => Option.none<number>()),
543 );
544 if (Option.isNone(status)) continue;
545 if (status.value === 200 || status.value === 503) {
546 const ready = status.value === 200;
547 if (ready !== lastReady) yield* Queue.offer(readinessChanges, undefined);
548 lastReady = ready;
549 }
550 if (status.value === 200) {
551 notReadyChecks = 0;
552 yield* Ref.set(notReadyChecksBeforeRestartRef, CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
553 continue;
554 }
555 if (status.value !== 503) continue;
556 notReadyChecks += 1;
557 const threshold = yield* Ref.get(notReadyChecksBeforeRestartRef);
558 if (notReadyChecks < threshold) continue;
559 notReadyChecks = 0;
560 yield* Ref.set(
561 notReadyChecksBeforeRestartRef,
562 Math.min(threshold * 2, CONNECTOR_NOT_READY_CHECKS_MAX),
563 );
564 yield* Effect.logWarning("Relay client has no tunnel connection; restarting it", {
565 pid: Number(connector.child.pid),
566 notReadyChecks: threshold,
567 tunnelId: connector.config.tunnelId,
568 tunnelName: connector.config.tunnelName,
569 });
570 yield* connector.child.kill().pipe(Effect.ignore);
571 }
572 });
573
574 const rememberHttp2Fallback = Effect.fnUntraced(function* () {
575 yield* Ref.set(http2Ref, true);
576 yield* secrets
577 .set(
578 HTTP2_FALLBACK_STATE,
579 new TextEncoder().encode(
580 encodeHttp2Fallback({ fellBackAt: yield* Clock.currentTimeMillis }),
581 ),
582 )
583 .pipe(
584 Effect.catch(() => Effect.logWarning("Could not persist the relay client HTTP/2 fallback")),
585 );
586 });
587
588 const switchToHttp2 = Effect.fnUntraced(function* (connector: ActiveConnector) {
589 const active = yield* Ref.get(activeRef);
590 const desiredConfig = yield* Ref.get(desiredConfigRef);
591 if ((yield* Ref.get(http2Ref)) || active !== connector || !desiredConfig) return;
592 yield* rememberHttp2Fallback();
593 yield* Effect.logWarning("Relay client keeps losing QUIC connections; switching to HTTP/2", {
594 pid: Number(connector.child.pid),
595 tunnelId: connector.config.tunnelId,
596 tunnelName: connector.config.tunnelName,
597 });
598 // Shutdown must not interrupt cleanup after active ownership is cleared.
599 yield* Effect.uninterruptible(stopActive);
600 const status = yield* reconcileConfig(desiredConfig);
601 // A failed spawn has no supervisor to request recovery.
602 if (status.status !== "running") {
603 yield* Queue.offer(recoveryRequests, desiredConfig);
604 }
605 }, reconcileSemaphore.withPermits(1));
606
607 const noteQuicStall = Effect.fnUntraced(function* (connector: ActiveConnector) {
608 if (transportChosenByUser || (yield* Ref.get(http2Ref))) return;
609 const now = yield* Clock.currentTimeMillis;
610 if ((yield* Ref.get(activeRef)) !== connector) return;
611 const stalls = yield* Ref.updateAndGet(quicStallsRef, (previous) => [
612 ...previous.filter((at) => now - at < QUIC_STALL_WINDOW_MS),
613 now,
614 ]);
615 if (stalls.length < QUIC_STALLS_BEFORE_HTTP2) return;
616 // Stopping the connector interrupts the output observer that called this.
617 yield* Effect.forkIn(switchToHttp2(connector), runtimeScope);
618 });
619
620 const observeConnectorOutput = (connector: ActiveConnector) => {
621 let rejectedRegistrations = 0;
622
623 return connector.child.all.pipe(
624 Stream.decodeText(),
625 Stream.splitLines,
626 Stream.map((line) => line.trim()),
627 Stream.filter((line) => line.length > 0),
628 Stream.runForEach((line) => {
629 const output = line.replaceAll(connector.config.connectorToken, "<redacted>");
630 const attributes = {
631 pid: Number(connector.child.pid),
632 tunnelId: connector.config.tunnelId,
633 tunnelName: connector.config.tunnelName,
634 output,
635 };
636 switch (classifyRelayClientOutput(line)) {
637 case "connected": {
638 rejectedRegistrations = 0;
639 const protocol = RELAY_CLIENT_PROTOCOL.exec(line)?.[1] ?? null;
640 return Deferred.succeed(connector.registered, undefined).pipe(
641 Effect.andThen(Ref.set(connector.protocol, protocol)),
642 // cloudflared started with `auto` fell back on its own; later starts skip QUIC too.
643 Effect.andThen(
644 protocol === "http2" && !transportChosenByUser
645 ? Ref.get(http2Ref).pipe(
646 Effect.flatMap((http2) => (http2 ? Effect.void : rememberHttp2Fallback())),
647 )
648 : Effect.void,
649 ),
650 Effect.andThen(
651 Effect.logInfo("Relay client tunnel connection registered", attributes),
652 ),
653 Effect.andThen(Queue.offer(tunnelConnections, undefined)),
654 Effect.andThen(Queue.offer(readinessChanges, undefined)),
655 Effect.andThen(
656 RelayClient.isPinnedManagedRelayClient(connector.executable)
657 ? Ref.getAndSet(prunedRef, true).pipe(
658 Effect.flatMap((pruned) =>
659 pruned ? Effect.void : relayClient.pruneManagedVersions,
660 ),
661 )
662 : Effect.void,
663 ),
664 Effect.asVoid,
665 );
666 }
667 case "unregistered":
668 return Effect.logInfo("Relay client tunnel connection unregistered", attributes);
669 case "warning":
670 if (/\btimeout: no recent network activity\b/u.test(line)) {
671 return noteQuicStall(connector).pipe(
672 Effect.andThen(
673 Effect.logWarning("Relay client reported a transport warning", attributes),
674 ),
675 );
676 }
677 if (isRejectedRelayClientTunnelOutput(line)) {
678 rejectedRegistrations += 1;
679 if (rejectedRegistrations >= TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY) {
680 rejectedRegistrations = 0;
681 return Effect.logWarning(
682 "Relay client tunnel was rejected; requesting recovery",
683 attributes,
684 ).pipe(
685 Effect.andThen(Queue.offer(recoveryRequests, connector.config)),
686 Effect.asVoid,
687 );
688 }
689 }
690 return Effect.logWarning("Relay client reported a transport warning", attributes);
691 case "debug": {
692 const metricsAddress = RELAY_CLIENT_METRICS_LINE.exec(line)?.[1];
693 return metricsAddress
694 ? Deferred.succeed(connector.metricsAddress, metricsAddress).pipe(
695 Effect.andThen(Effect.logDebug("Relay client output", attributes)),
696 )
697 : Effect.logDebug("Relay client output", attributes);
698 }
699 }
700 }),
701 Effect.catchCause((cause) =>
702 Effect.logWarning("Relay client output observer failed", {
703 cause,
704 pid: Number(connector.child.pid),
705 tunnelId: connector.config.tunnelId,
706 tunnelName: connector.config.tunnelName,
707 }),
708 ),
709 );
710 };
711
712 reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) {
713 if (!config || config.providerKind !== "cloudflare_tunnel") {
714 yield* stopActive;
715 return config
716 ? { status: "unsupported", providerKind: config.providerKind }
717 : { status: "disabled" };
718 }
719
720 const nextConfigKey = runtimeConfigKey(config);
721 const active = yield* Ref.get(activeRef);
722 if (active?.configKey === nextConfigKey) {
723 const isRunning = yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false));
724 if (isRunning) {
725 return {
726 status: "running",
727 providerKind: "cloudflare_tunnel",
728 pid: Number(active.child.pid),
729 ...(active.config.tunnelId ? { tunnelId: active.config.tunnelId } : {}),
730 ...(active.config.tunnelName ? { tunnelName: active.config.tunnelName } : {}),
731 } satisfies CloudManagedEndpointRuntimeStatus;
732 }
733 }
734
735 yield* stopActive;
736
737 const executable = yield* relayClient.resolve;
738 if (needsPinnedRelayClient(executable)) {
739 yield* ensurePinnedRelayClient;
740 yield* startPinnedRelayClientRetries;
741 }
742 if (executable.status !== "available") {
743 return {
744 status: "failed",
745 providerKind: "cloudflare_tunnel",
746 failure: executable.status === "unsupported" ? "unsupported-platform" : "not-installed",
747 reason:
748 executable.status === "unsupported"
749 ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.`
750 : "The relay client is not installed.",
751 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
752 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
753 } satisfies CloudManagedEndpointRuntimeStatus;
754 }
755
756 const http2 = yield* Ref.get(http2Ref);
757 const connectorScope = yield* Scope.make("sequential");
758 const child = yield* spawner
759 .spawn(
760 ChildProcess.make(
761 executable.executablePath,
762 [
763 "tunnel",
764 "--no-autoupdate",
765 "--loglevel",
766 "info",
767 "--output",
768 "default",
769 "--metrics",
770 "127.0.0.1:0",
771 ...(http2 ? ["--protocol", "http2"] : []),
772 "run",
773 ],
774 {
775 detached: false,
776 env: {
777 ...env,
778 TUNNEL_TOKEN: config.connectorToken,
779 },
780 shell: false,
781 stderr: "pipe",
782 stdout: "pipe",
783 },
784 ),
785 )
786 .pipe(
787 Effect.provideService(Scope.Scope, connectorScope),
788 Effect.tap((child) =>
789 Effect.logInfo("Relay client process started; waiting for tunnel connection", {
790 pid: Number(child.pid),
791 source: executable.source,
792 version: executable.version,
793 protocol: http2 ? "http2" : "auto",
794 tunnelId: config.tunnelId,
795 tunnelName: config.tunnelName,
796 }),
797 ),
798 Effect.catch((cause) =>
799 Effect.logWarning("Failed to start relay client", {
800 cause,
801 tunnelId: config.tunnelId,
802 tunnelName: config.tunnelName,
803 }).pipe(
804 Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)),
805 Effect.as({
806 status: "failed",
807 providerKind: "cloudflare_tunnel",
808 failure: "spawn-failed",
809 reason: String(cause),
810 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
811 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
812 } satisfies CloudManagedEndpointRuntimeStatus),
813 ),
814 ),
815 );
816
817 if ("status" in child && child.status === "failed") {
818 return child;
819 }
820
821 if (!("status" in child)) {
822 const connector = {
823 child,
824 executable,
825 scope: connectorScope,
826 configKey: nextConfigKey,
827 config,
828 startedAtMillis: yield* Clock.currentTimeMillis,
829 registered: yield* Deferred.make<void>(),
830 metricsAddress: yield* Deferred.make<string>(),
831 protocol: yield* Ref.make<string | null>(null),
832 } satisfies ActiveConnector;
833 yield* Ref.set(activeRef, connector);
834 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
835 yield* Effect.forkIn(superviseConnector(connector), connectorScope);
836 yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope);
837 yield* Effect.forkIn(watchConnectorReadiness(connector), connectorScope);
838 return {
839 status: "running",
840 providerKind: "cloudflare_tunnel",
841 pid: Number(child.pid),
842 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
843 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
844 } satisfies CloudManagedEndpointRuntimeStatus;
845 }
846
847 return {
848 status: "failed",
849 providerKind: "cloudflare_tunnel",
850 failure: "spawn-failed",
851 reason: "Relay client did not start.",
852 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
853 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
854 } satisfies CloudManagedEndpointRuntimeStatus;
855 });
856
857 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(
858 (requested: RelayManagedEndpointRuntimeConfig | null) =>
859 reconcileSemaphore.withPermits(1)(
860 Effect.gen(function* () {
861 const config = connectAllowed ? requested : null;
862 // Keep backoff when recovery returns the same tunnel/token, or every recovery respawns a crash loop.
863 const desired = yield* Ref.get(desiredConfigRef);
864 const unchanged =
865 desired !== null &&
866 config !== null &&
867 runtimeConfigKey(desired) === runtimeConfigKey(config);
868 if (!unchanged) {
869 yield* Ref.set(restartDelayRef, 0);
870 }
871 yield* Ref.set(desiredConfigRef, config);
872 return yield* reconcileConfig(config);
873 }),
874 ),
875 withRelayClientLog,
876 );
877
878 const restartConnector = Effect.fn("CloudManagedEndpointRuntime.restartConnector")(function* (
879 pid: number,
880 stillDead: Effect.Effect<boolean>,
881 ) {
882 const active = yield* Ref.get(activeRef);
883 if (!active || Number(active.child.pid) !== pid || !(yield* stillDead)) return false;
884 yield* Effect.logWarning(
885 "Cloudflare has no live connection from the relay client; restarting it",
886 { pid, tunnelId: active.config.tunnelId, tunnelName: active.config.tunnelName },
887 );
888 // The supervisor starts the replacement with its usual crash-loop backoff.
889 return yield* active.child.kill().pipe(
890 Effect.as(true),
891 Effect.orElseSucceed(() => false),
892 );
893 }, reconcileSemaphore.withPermits(1));
894
895 const readMetrics = <S extends Schema.Decoder<unknown>>(url: string, schema: S) =>
896 httpClient.get(url).pipe(
897 Effect.flatMap(HttpClientResponse.schemaBodyJson(schema)),
898 Effect.scoped,
899 Effect.withTracerEnabled(false),
900 Effect.timeoutOption(CONNECTOR_READINESS_REQUEST_TIMEOUT),
901 Effect.map(Option.getOrNull),
902 Effect.orElseSucceed(() => null),
903 );
904
905 const inspectConnector = Effect.gen(function* () {
906 const active = yield* Ref.get(activeRef);
907 if (!active || !(yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false)))) {
908 return null;
909 }
910 const pid = Number(active.child.pid);
911 const protocol = yield* Ref.get(active.protocol);
912 if (!(yield* Deferred.isDone(active.metricsAddress))) {
913 return { pid, readyConnections: null, hostname: null, protocol } satisfies ConnectorFacts;
914 }
915 const metricsUrl = `http://${yield* Deferred.await(active.metricsAddress)}`;
916 const [ready, config] = yield* Effect.all(
917 [
918 readMetrics(`${metricsUrl}/ready`, ConnectorReadyBody),
919 readMetrics(`${metricsUrl}/config`, ConnectorConfigBody),
920 ],
921 { concurrency: 2 },
922 );
923 return {
924 pid,
925 readyConnections: ready?.readyConnections ?? null,
926 hostname: config?.config.ingress?.find((rule) => rule.hostname)?.hostname ?? null,
927 protocol,
928 } satisfies ConnectorFacts;
929 });
930
931 const runtime = CloudManagedEndpointRuntime.of({
932 applyConfig,
933 inspectConnector,
934 readinessChanged: Stream.fromQueue(readinessChanges),
935 recoveryRequests: Stream.fromQueue(recoveryRequests),
936 requestRecovery: Effect.fn("CloudManagedEndpointRuntime.requestRecovery")(function (config) {
937 return Queue.offer(recoveryRequests, config).pipe(Effect.asVoid);
938 }),
939 restartConnector,
940 tunnelConnected: Stream.fromQueue(tunnelConnections),
941 withLinkStateLock: linkStateSemaphore.withPermits(1),
942 });
943
944 yield* Effect.addFinalizer(() => runtime.applyConfig(null));
945 return runtime;
946});
947
948export const layer = Layer.effect(CloudManagedEndpointRuntime, make);