1import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
2import * as RelayClient from "@t3tools/shared/relayClient";
3import * as Clock from "effect/Clock";
4import * as Context from "effect/Context";
5import * as Deferred from "effect/Deferred";
6import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
11import * as Queue from "effect/Queue";
12import * as Ref from "effect/Ref";
13import * as Result from "effect/Result";
14import * as Semaphore from "effect/Semaphore";
15import * as Scope from "effect/Scope";
16import * as Stream from "effect/Stream";
17import { HttpClient } from "effect/http";
18import * as ChildProcess from "effect/process/ChildProcess";
19import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
20
21export type CloudManagedEndpointRuntimeStatus =
22 | {
23 readonly status: "disabled";
24 }
25 | {
26 readonly status: "failed";
27 readonly providerKind: RelayManagedEndpointRuntimeConfig["providerKind"];
28 readonly failure: "unsupported-platform" | "not-installed" | "spawn-failed";
29 readonly reason: string;
30 readonly tunnelId?: string;
31 readonly tunnelName?: string;
32 }
33 | {
34 readonly status: "running";
35 readonly providerKind: "cloudflare_tunnel";
36 readonly pid: number;
37 readonly tunnelId?: string;
38 readonly tunnelName?: string;
39 }
40 | {
41 readonly status: "unsupported";
42 readonly providerKind: RelayManagedEndpointRuntimeConfig["providerKind"];
43 };
44
45export class CloudManagedEndpointRuntime extends Context.Service<
46 CloudManagedEndpointRuntime,
47 {
48 readonly applyConfig: (
49 config: RelayManagedEndpointRuntimeConfig | null,
50 ) => Effect.Effect<CloudManagedEndpointRuntimeStatus>;
51 readonly recoveryRequests: Stream.Stream<RelayManagedEndpointRuntimeConfig>;
52 readonly requestRecovery: (config: RelayManagedEndpointRuntimeConfig) => Effect.Effect<void>;
53 /** Emits when the connector registers a tunnel connection, i.e. the relay can reach us again. */
54 readonly tunnelConnected: Stream.Stream<void>;
55 readonly withLinkStateLock: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
56 }
57>()("t3/cloud/ManagedEndpointRuntime/CloudManagedEndpointRuntime") {}
58
59interface ActiveConnector {
60 readonly child: ChildProcessSpawner.ChildProcessHandle;
61 readonly executable: RelayClient.AvailableRelayClient;
62 readonly scope: Scope.Closeable;
63 readonly configKey: string;
64 readonly config: RelayManagedEndpointRuntimeConfig;
65 readonly startedAtMillis: number;
66 /** Completes when the connector first registers a tunnel connection. */
67 readonly registered: Deferred.Deferred<void>;
68 /** Completes with the loopback address of the connector's metrics server. */
69 readonly metricsAddress: Deferred.Deferred<string>;
70}
71
72// A connector that exits before running this long is treated as part of a
73// crash loop; one that stays up at least this long earns an immediate restart
74// again. Without the backoff below, a relay client that fails instantly (a
75// stale version-manager shim, a bad binary) respawns ~100 times per second
76// until the accumulated tracing exhausts the V8 heap.
77const RELAY_RESTART_STABLE_UPTIME_MS = 30_000;
78const RELAY_RESTART_BACKOFF_BASE_MS = 1_000;
79const RELAY_RESTART_BACKOFF_MAX_MS = 60_000;
80// Newly created tunnels can fail authorization briefly while Cloudflare propagates their token.
81const TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY = 4;
82// A connector that never registers a connection may hold a token for a tunnel
83// that no longer exists, rejected in wording the output check does not know.
84// Ask for recovery after this long, and again at this interval while it stays
85// unconnected; the relay hands back the same tunnel when it is still live.
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;
99// A failed background install of the pinned relay client is retried no sooner
100// than this, so a crash-looping connector or an offline host does not redownload
101// on every reconcile.
102const RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS = 10 * 60_000;
103
104/**
105 * A linked host converges on the pinned managed release when it has no relay
106 * client or runs another managed release. A PATH binary or an explicit override
107 * is the user's choice, and the CLI asks before downloading, so both stay put.
108 */
109function needsPinnedRelayClient(executable: RelayClient.RelayClientStatus): boolean {
110 return (
111 executable.status === "missing" ||
112 (executable.status === "available" &&
113 executable.source === "managed" &&
114 !RelayClient.isPinnedManagedRelayClient(executable))
115 );
116}
117
118export function classifyRelayClientOutput(line: string): "connected" | "warning" | "debug" {
119 if (/\bRegistered tunnel connection\b/iu.test(line)) {
120 return "connected";
121 }
122 // cloudflared uses zerolog level tokens. FTL (fatal) and PNC (panic) are more
123 // severe than ERR, so they must surface at least as loudly — without them a
124 // fatal connector failure would be logged at debug and hidden.
125 return /\b(?:ERR|WRN|FTL|PNC)\b/u.test(line) ? "warning" : "debug";
126}
127
128/**
129 * Cloudflare's edge rejects a connector whose tunnel was deleted or whose
130 * token no longer matches. The edge words this differently over time
131 * (`Failed to get tunnel`, `Tunnel not found`, ...), sometimes prefixed with
132 * `Unauthorized:`. Treat any `Unauthorized:` registration error as a rejection,
133 * plus the unprefixed messages seen so far. Transient rejections while a new
134 * tunnel's token propagates are absorbed by requiring several in a row.
135 */
136export function isRejectedRelayClientTunnelOutput(line: string): boolean {
137 return (
138 /\bRegister tunnel error from server side\b/iu.test(line) &&
139 /error="(?:Unauthorized:[^"]*|Failed to get tunnel|Tunnel not found|Record for tunnel not found|Invalid tunnel secret)"/iu.test(
140 line,
141 )
142 );
143}
144
145/** Connector startup failures can clear after installation or a later spawn attempt. */
146export function isRetryableManagedEndpointRuntimeStatus(status: unknown): boolean {
147 if (typeof status !== "object" || status === null || !("status" in status)) {
148 return false;
149 }
150 if (status.status !== "failed" || !("failure" in status)) {
151 return false;
152 }
153 return status.failure === "not-installed" || status.failure === "spawn-failed";
154}
155
156function runtimeConfigKey(config: RelayManagedEndpointRuntimeConfig): string {
157 return JSON.stringify({
158 providerKind: config.providerKind,
159 connectorToken: config.connectorToken,
160 tunnelId: config.tunnelId ?? null,
161 tunnelName: config.tunnelName ?? null,
162 });
163}
164
165const stopConnector = (connector: ActiveConnector | null) =>
166 connector
167 ? Scope.close(connector.scope, Exit.void).pipe(
168 Effect.tap(() =>
169 Effect.logInfo("Relay client stopped", {
170 pid: Number(connector.child.pid),
171 }),
172 ),
173 Effect.ignore,
174 )
175 : Effect.void;
176
177/** @public Service construction is part of the canonical Effect module API. */
178export const make = Effect.gen(function* () {
179 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
180 const relayClient = yield* RelayClient.RelayClient;
181 const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
182 const activeRef = yield* Ref.make<ActiveConnector | null>(null);
183 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
184 const recoveryRequests = yield* Queue.sliding<RelayManagedEndpointRuntimeConfig>(1);
185 const tunnelConnections = yield* Queue.sliding<void>(1);
186 const reconcileSemaphore = yield* Semaphore.make(1);
187 const restartDelayRef = yield* Ref.make(0);
188 const linkStateSemaphore = yield* Semaphore.make(1);
189 const runtimeScope = yield* Effect.scope;
190 const installInFlightRef = yield* Ref.make(false);
191 const lastInstallFailureAtRef = yield* Ref.make<number | null>(null);
192 const prunedRef = yield* Ref.make(false);
193 const retryLoopRunningRef = yield* Ref.make(false);
194 const notReadyChecksBeforeRestartRef = yield* Ref.make(CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
195 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
196
197 const stopActive = Effect.gen(function* () {
198 const active = yield* Ref.getAndSet(activeRef, null);
199 yield* stopConnector(active);
200 });
201
202 const superviseConnector = (connector: ActiveConnector) =>
203 Effect.gen(function* () {
204 const result = yield* Effect.result(connector.child.exitCode);
205 const activeAtExit = yield* Ref.get(activeRef);
206 if (
207 activeAtExit?.child.pid !== connector.child.pid ||
208 activeAtExit.configKey !== connector.configKey
209 ) {
210 return;
211 }
212 const uptimeMillis = (yield* Clock.currentTimeMillis) - connector.startedAtMillis;
213 // The first crash restarts immediately; every further crash inside the
214 // stable-uptime window doubles the wait, up to the cap. The delay runs
215 // before the semaphore so a user config change is never blocked behind
216 // it, and reconcileConfig re-checks the desired config afterwards.
217 const restartDelayMillis = yield* Ref.modify(restartDelayRef, (current) => {
218 if (uptimeMillis >= RELAY_RESTART_STABLE_UPTIME_MS) {
219 return [0, 0];
220 }
221 return [
222 current,
223 current === 0
224 ? RELAY_RESTART_BACKOFF_BASE_MS
225 : Math.min(current * 2, RELAY_RESTART_BACKOFF_MAX_MS),
226 ];
227 });
228 if (restartDelayMillis > 0) {
229 yield* Effect.logWarning("Relay client is crash-looping; delaying restart", {
230 pid: Number(connector.child.pid),
231 uptimeMillis,
232 restartDelayMillis,
233 tunnelId: connector.config.tunnelId,
234 tunnelName: connector.config.tunnelName,
235 });
236 yield* Effect.sleep(Duration.millis(restartDelayMillis));
237 }
238 yield* reconcileSemaphore.withPermits(1)(
239 Effect.gen(function* () {
240 const active = yield* Ref.get(activeRef);
241 if (
242 active?.child.pid !== connector.child.pid ||
243 active.configKey !== connector.configKey
244 ) {
245 return;
246 }
247 yield* Ref.set(activeRef, null);
248 yield* stopConnector(connector);
249
250 const desiredConfig = yield* Ref.get(desiredConfigRef);
251 if (
252 !desiredConfig ||
253 desiredConfig.providerKind !== "cloudflare_tunnel" ||
254 runtimeConfigKey(desiredConfig) !== connector.configKey
255 ) {
256 return;
257 }
258
259 yield* Effect.logWarning("Relay client exited; restarting", {
260 pid: Number(connector.child.pid),
261 ...(Result.isSuccess(result)
262 ? { exitCode: Number(result.success) }
263 : { cause: result.failure }),
264 tunnelId: connector.config.tunnelId,
265 tunnelName: connector.config.tunnelName,
266 });
267 yield* Queue.offer(recoveryRequests, connector.config);
268 yield* reconcileConfig(desiredConfig);
269 }),
270 );
271 }).pipe(
272 Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })),
273 );
274
275 // Installs the pinned relay client in the background, then moves the connector
276 // to it. The new connector starts next to the old one on the same tunnel, and
277 // the old one stops only once the new one registers, so clients never see the
278 // relay lose this host during the swap.
279 const installPinnedRelayClient = Effect.gen(function* () {
280 if (yield* Ref.getAndSet(installInFlightRef, true)) return;
281 yield* Effect.logInfo("Installing the pinned relay client", {
282 version: RelayClient.CLOUDFLARED_VERSION,
283 });
284 yield* relayClient.install.pipe(
285 Effect.tap(() => Ref.set(lastInstallFailureAtRef, null)),
286 Effect.flatMap((installed) =>
287 reconcileSemaphore.withPermits(1)(
288 Effect.gen(function* () {
289 const desiredConfig = yield* Ref.get(desiredConfigRef);
290 if (!desiredConfig || desiredConfig.providerKind !== "cloudflare_tunnel") return;
291 const active = yield* Ref.get(activeRef);
292 // A self-updated binary in the pinned folder shares the installed path,
293 // so the version decides whether the running connector is current.
294 if (
295 active?.executable.executablePath === installed.executablePath &&
296 active.executable.version === installed.version
297 ) {
298 return;
299 }
300 // A fresh binary can fail its probe while a scanner holds it; restarting
301 // then would land on the same older binary, so leave it to the retry.
302 const resolved = yield* relayClient.resolve;
303 if (
304 resolved.status !== "available" ||
305 resolved.executablePath !== installed.executablePath ||
306 resolved.version !== installed.version
307 ) {
308 return;
309 }
310 yield* Effect.logInfo("Relay client installed; moving the connector to it", {
311 version: installed.version,
312 previousVersion: active?.executable.version,
313 });
314 // Detach the old connector so the reconcile starts a new one beside it.
315 const previous = yield* Ref.getAndSet(activeRef, null);
316 const status = yield* reconcileConfig(desiredConfig);
317 const next = yield* Ref.get(activeRef);
318 if (!next) {
319 yield* Effect.logWarning("Relay client did not start after the update", status);
320 // A still-running old connector keeps serving and the retry loop
321 // tries the pin again. One that exited while detached was skipped
322 // by its supervisor, so ask for recovery like an exited connector.
323 const previousRunning = previous
324 ? yield* previous.child.isRunning.pipe(Effect.orElseSucceed(() => false))
325 : false;
326 if (previous && previousRunning) {
327 yield* Ref.set(activeRef, previous);
328 } else {
329 yield* stopConnector(previous);
330 yield* Queue.offer(recoveryRequests, desiredConfig);
331 }
332 return;
333 }
334 // Runs in the new connector's scope: the old one stops when the new
335 // one registers, or as soon as the new one is stopped or replaced.
336 if (previous) {
337 yield* Deferred.await(next.registered).pipe(
338 Effect.ensuring(stopConnector(previous)),
339 Effect.forkIn(next.scope),
340 );
341 }
342 }),
343 ),
344 ),
345 Effect.catchTags({
346 RelayClientInstallError: (error) =>
347 Clock.currentTimeMillis.pipe(
348 Effect.flatMap((failedAt) => Ref.set(lastInstallFailureAtRef, failedAt)),
349 Effect.andThen(
350 Effect.logWarning("Could not install the pinned relay client", {
351 reason: error.reason,
352 message: error.message,
353 cause: error.cause,
354 }),
355 ),
356 ),
357 }),
358 Effect.ensuring(Ref.set(installInFlightRef, false)),
359 );
360 });
361
362 // Reconciles call this; a recent failure skips the attempt so a crash-looping
363 // connector or an offline host does not redownload on every restart.
364 const ensurePinnedRelayClient = Effect.gen(function* () {
365 const now = yield* Clock.currentTimeMillis;
366 const lastFailureAt = yield* Ref.get(lastInstallFailureAtRef);
367 if (lastFailureAt !== null && now - lastFailureAt < RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS) {
368 return;
369 }
370 yield* installPinnedRelayClient;
371 }).pipe(Effect.forkIn(runtimeScope), Effect.asVoid);
372
373 // Until the host runs the pinned release, retry the install every interval.
374 // applyConfig returns early while an older connector is healthy, and some
375 // activations never retry a missing client, so one runtime loop covers both.
376 // It paces itself, so it skips the failure gate above.
377 const retryPinnedRelayClientInstall = Effect.gen(function* () {
378 while (true) {
379 yield* Effect.sleep(Duration.millis(RELAY_CLIENT_INSTALL_RETRY_INTERVAL_MS));
380 const desiredConfig = yield* Ref.get(desiredConfigRef);
381 if (!desiredConfig || desiredConfig.providerKind !== "cloudflare_tunnel") return;
382 // Stop only once a connector runs on the pin. The pin can be on disk while
383 // an older connector still runs, or while none runs because a post-install
384 // probe failed; another install call then finds it and starts the connector.
385 const active = yield* Ref.get(activeRef);
386 if (active && !needsPinnedRelayClient(active.executable)) return;
387 yield* installPinnedRelayClient;
388 }
389 }).pipe(Effect.ensuring(Ref.set(retryLoopRunningRef, false)));
390
391 const startPinnedRelayClientRetries = Effect.gen(function* () {
392 if (yield* Ref.getAndSet(retryLoopRunningRef, true)) return;
393 yield* Effect.forkIn(retryPinnedRelayClientInstall, runtimeScope);
394 });
395
396 // Requests recovery while the connector has not registered a connection,
397 // once per timeout, until it connects or is replaced.
398 const watchConnectorRegistration = (connector: ActiveConnector) =>
399 Effect.gen(function* () {
400 yield* Effect.sleep(CONNECTOR_REGISTRATION_TIMEOUT);
401 if (yield* Deferred.isDone(connector.registered)) return true;
402 yield* Effect.logWarning(
403 "Relay client has not registered a tunnel connection; requesting recovery",
404 {
405 pid: Number(connector.child.pid),
406 tunnelId: connector.config.tunnelId,
407 tunnelName: connector.config.tunnelName,
408 },
409 );
410 yield* Queue.offer(recoveryRequests, connector.config);
411 return false;
412 }).pipe(Effect.repeat({ until: (connected) => connected }));
413
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
456 const observeConnectorOutput = (connector: ActiveConnector) => {
457 let rejectedRegistrations = 0;
458
459 return connector.child.all.pipe(
460 Stream.decodeText(),
461 Stream.splitLines,
462 Stream.map((line) => line.trim()),
463 Stream.filter((line) => line.length > 0),
464 Stream.runForEach((line) => {
465 const output = line.replaceAll(connector.config.connectorToken, "<redacted>");
466 const attributes = {
467 pid: Number(connector.child.pid),
468 tunnelId: connector.config.tunnelId,
469 tunnelName: connector.config.tunnelName,
470 output,
471 };
472 switch (classifyRelayClientOutput(line)) {
473 case "connected":
474 rejectedRegistrations = 0;
475 return Deferred.succeed(connector.registered, undefined).pipe(
476 Effect.andThen(
477 Effect.logInfo("Relay client tunnel connection registered", attributes),
478 ),
479 Effect.andThen(Queue.offer(tunnelConnections, undefined)),
480 Effect.andThen(
481 RelayClient.isPinnedManagedRelayClient(connector.executable)
482 ? Ref.getAndSet(prunedRef, true).pipe(
483 Effect.flatMap((pruned) =>
484 pruned ? Effect.void : relayClient.pruneManagedVersions,
485 ),
486 )
487 : Effect.void,
488 ),
489 Effect.asVoid,
490 );
491 case "warning":
492 if (isRejectedRelayClientTunnelOutput(line)) {
493 rejectedRegistrations += 1;
494 if (rejectedRegistrations >= TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY) {
495 rejectedRegistrations = 0;
496 return Effect.logWarning(
497 "Relay client tunnel was rejected; requesting recovery",
498 attributes,
499 ).pipe(
500 Effect.andThen(Queue.offer(recoveryRequests, connector.config)),
501 Effect.asVoid,
502 );
503 }
504 }
505 return Effect.logWarning("Relay client reported a transport warning", attributes);
506 case "debug": {
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 }
514 }
515 }),
516 Effect.catchCause((cause) =>
517 Effect.logWarning("Relay client output observer failed", {
518 cause,
519 pid: Number(connector.child.pid),
520 tunnelId: connector.config.tunnelId,
521 tunnelName: connector.config.tunnelName,
522 }),
523 ),
524 );
525 };
526
527 reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) {
528 if (!config || config.providerKind !== "cloudflare_tunnel") {
529 yield* stopActive;
530 return config
531 ? { status: "unsupported", providerKind: config.providerKind }
532 : { status: "disabled" };
533 }
534
535 const nextConfigKey = runtimeConfigKey(config);
536 const active = yield* Ref.get(activeRef);
537 if (active?.configKey === nextConfigKey) {
538 const isRunning = yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false));
539 if (isRunning) {
540 return {
541 status: "running",
542 providerKind: "cloudflare_tunnel",
543 pid: Number(active.child.pid),
544 ...(active.config.tunnelId ? { tunnelId: active.config.tunnelId } : {}),
545 ...(active.config.tunnelName ? { tunnelName: active.config.tunnelName } : {}),
546 } satisfies CloudManagedEndpointRuntimeStatus;
547 }
548 }
549
550 yield* stopActive;
551
552 const executable = yield* relayClient.resolve;
553 if (needsPinnedRelayClient(executable)) {
554 yield* ensurePinnedRelayClient;
555 yield* startPinnedRelayClientRetries;
556 }
557 if (executable.status !== "available") {
558 return {
559 status: "failed",
560 providerKind: "cloudflare_tunnel",
561 failure: executable.status === "unsupported" ? "unsupported-platform" : "not-installed",
562 reason:
563 executable.status === "unsupported"
564 ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.`
565 : "The relay client is not installed.",
566 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
567 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
568 } satisfies CloudManagedEndpointRuntimeStatus;
569 }
570
571 const connectorScope = yield* Scope.make("sequential");
572 const child = yield* spawner
573 .spawn(
574 ChildProcess.make(
575 executable.executablePath,
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 ],
587 {
588 detached: false,
589 env: {
590 ...process.env,
591 TUNNEL_TOKEN: config.connectorToken,
592 },
593 shell: false,
594 stderr: "pipe",
595 stdout: "pipe",
596 },
597 ),
598 )
599 .pipe(
600 Effect.provideService(Scope.Scope, connectorScope),
601 Effect.tap((child) =>
602 Effect.logInfo("Relay client process started; waiting for tunnel connection", {
603 pid: Number(child.pid),
604 source: executable.source,
605 version: executable.version,
606 tunnelId: config.tunnelId,
607 tunnelName: config.tunnelName,
608 }),
609 ),
610 Effect.catch((cause) =>
611 Effect.logWarning("Failed to start relay client", {
612 cause,
613 tunnelId: config.tunnelId,
614 tunnelName: config.tunnelName,
615 }).pipe(
616 Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)),
617 Effect.as({
618 status: "failed",
619 providerKind: "cloudflare_tunnel",
620 failure: "spawn-failed",
621 reason: String(cause),
622 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
623 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
624 } satisfies CloudManagedEndpointRuntimeStatus),
625 ),
626 ),
627 );
628
629 if ("status" in child && child.status === "failed") {
630 return child;
631 }
632
633 if (!("status" in child)) {
634 const connector = {
635 child,
636 executable,
637 scope: connectorScope,
638 configKey: nextConfigKey,
639 config,
640 startedAtMillis: yield* Clock.currentTimeMillis,
641 registered: yield* Deferred.make<void>(),
642 metricsAddress: yield* Deferred.make<string>(),
643 } satisfies ActiveConnector;
644 yield* Ref.set(activeRef, connector);
645 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
646 yield* Effect.forkIn(superviseConnector(connector), connectorScope);
647 yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope);
648 yield* Effect.forkIn(watchConnectorReadiness(connector), connectorScope);
649 return {
650 status: "running",
651 providerKind: "cloudflare_tunnel",
652 pid: Number(child.pid),
653 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
654 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
655 } satisfies CloudManagedEndpointRuntimeStatus;
656 }
657
658 return {
659 status: "failed",
660 providerKind: "cloudflare_tunnel",
661 failure: "spawn-failed",
662 reason: "Relay client did not start.",
663 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
664 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
665 } satisfies CloudManagedEndpointRuntimeStatus;
666 });
667
668 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(
669 (config: RelayManagedEndpointRuntimeConfig | null) =>
670 reconcileSemaphore.withPermits(1)(
671 Effect.gen(function* () {
672 // A real config change starts over with a fresh backoff. Recovery
673 // that hands back the same tunnel and token must keep the delay, or
674 // a crash-looping connector respawns on every recovery round trip.
675 const desired = yield* Ref.get(desiredConfigRef);
676 const unchanged =
677 desired !== null &&
678 config !== null &&
679 runtimeConfigKey(desired) === runtimeConfigKey(config);
680 if (!unchanged) {
681 yield* Ref.set(restartDelayRef, 0);
682 }
683 yield* Ref.set(desiredConfigRef, config);
684 return yield* reconcileConfig(config);
685 }),
686 ),
687 );
688
689 const runtime = CloudManagedEndpointRuntime.of({
690 applyConfig,
691 recoveryRequests: Stream.fromQueue(recoveryRequests),
692 requestRecovery: (config) => Queue.offer(recoveryRequests, config).pipe(Effect.asVoid),
693 tunnelConnected: Stream.fromQueue(tunnelConnections),
694 withLinkStateLock: linkStateSemaphore.withPermits(1),
695 });
696
697 yield* Effect.addFinalizer(() => runtime.applyConfig(null));
698 return runtime;
699});
700
701export const layer = Layer.effect(CloudManagedEndpointRuntime, make);