fix(connect): remove tunnels after hosts go offline

apps/server/src/cloud/ManagedEndpointRuntime.test.ts

@@ -6,9 +6,11 @@ import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as PlatformError from "effect/PlatformError";
+import * as Queue from "effect/Queue";
 import * as Sink from "effect/Sink";
 import * as Stream from "effect/Stream";
 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
+import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
 import * as RelayClient from "@t3tools/shared/relayClient";
 
 import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
@@ -60,6 +62,7 @@ function makeHandle(input: {
   readonly onKill: () => void;
   readonly isRunning?: () => boolean;
   readonly exitCode?: Effect.Effect<ChildProcessSpawner.ExitCode>;
+  readonly output?: Stream.Stream<Uint8Array>;
 }) {
   return ChildProcessSpawner.makeHandle({
     pid: ChildProcessSpawner.ProcessId(input.pid),
@@ -73,13 +76,70 @@ function makeHandle(input: {
     stdin: Sink.drain,
     stdout: Stream.empty,
     stderr: Stream.empty,
-    all: Stream.empty,
+    all: input.output ?? Stream.empty,
     getInputFd: () => Sink.drain,
     getOutputFd: () => Stream.empty,
   });
 }
 
 describe("CloudManagedEndpointRuntime", () => {
+  it("retries connector startup failures but stops for unsupported runtimes", () => {
+    expect(
+      ManagedEndpointRuntime.isRetryableManagedEndpointRuntimeStatus({
+        status: "failed",
+        failure: "not-installed",
+        reason: "The relay client is not installed.",
+      }),
+    ).toBe(true);
+    expect(
+      ManagedEndpointRuntime.isRetryableManagedEndpointRuntimeStatus({
+        status: "failed",
+        failure: "spawn-failed",
+        reason: "spawn failed",
+      }),
+    ).toBe(true);
+    expect(
+      ManagedEndpointRuntime.isRetryableManagedEndpointRuntimeStatus({
+        status: "failed",
+        failure: "unsupported-platform",
+        reason: "Relay client is unsupported on linux-arm.",
+      }),
+    ).toBe(false);
+    expect(
+      ManagedEndpointRuntime.isRetryableManagedEndpointRuntimeStatus({ status: "unsupported" }),
+    ).toBe(false);
+  });
+
+  it.effect("serializes updates to persisted cloud link state", () =>
+    Effect.gen(function* () {
+      const firstEntered = yield* Deferred.make<void>();
+      const releaseFirst = yield* Deferred.make<void>();
+      const secondEntered = yield* Deferred.make<void>();
+      const runtime = yield* buildCloudManagedEndpointRuntime(
+        ChildProcessSpawner.make(() => Effect.die("unused")),
+      );
+
+      const first = yield* runtime
+        .withLinkStateLock(
+          Deferred.succeed(firstEntered, undefined).pipe(
+            Effect.andThen(Deferred.await(releaseFirst)),
+          ),
+        )
+        .pipe(Effect.forkChild);
+      yield* Deferred.await(firstEntered);
+
+      const second = yield* runtime
+        .withLinkStateLock(Deferred.succeed(secondEntered, undefined))
+        .pipe(Effect.forkChild);
+      expect(yield* Deferred.isDone(secondEntered)).toBe(false);
+
+      yield* Deferred.succeed(releaseFirst, undefined);
+      yield* Fiber.join(first);
+      yield* Fiber.join(second);
+      expect(yield* Deferred.isDone(secondEntered)).toBe(true);
+    }),
+  );
+
   it("classifies Cloudflare connection and warning output", () => {
     expect(
       ManagedEndpointRuntime.classifyRelayClientOutput(
@@ -107,6 +167,125 @@ describe("CloudManagedEndpointRuntime", () => {
     ).toBe("warning");
   });
 
+  it("recognizes tunnel authorization failures without matching ordinary transport errors", () => {
+    expect(
+      ManagedEndpointRuntime.isRejectedRelayClientTunnelOutput(
+        '2026-06-17T02:00:00Z ERR Register tunnel error from server side error="Unauthorized: Failed to get tunnel" connIndex=0',
+      ),
+    ).toBe(true);
+    expect(
+      ManagedEndpointRuntime.isRejectedRelayClientTunnelOutput(
+        '2026-06-17T02:00:00Z ERR Register tunnel error from server side error="Unauthorized: Record for tunnel not found" connIndex=0',
+      ),
+    ).toBe(true);
+    expect(
+      ManagedEndpointRuntime.isRejectedRelayClientTunnelOutput(
+        '2026-06-17T02:00:00Z ERR Register tunnel error from server side error="Unauthorized: Invalid tunnel secret" connIndex=0',
+      ),
+    ).toBe(true);
+    expect(
+      ManagedEndpointRuntime.isRejectedRelayClientTunnelOutput(
+        '2026-06-17T02:00:00Z ERR Register tunnel error from server side error="connection timed out" connIndex=0',
+      ),
+    ).toBe(false);
+  });
+
+  it.effect("keeps recovery requests sent before the server starts consuming them", () =>
+    Effect.gen(function* () {
+      const runtime = yield* buildCloudManagedEndpointRuntime(
+        ChildProcessSpawner.make(() => Effect.die("unused")),
+      );
+      const config = {
+        providerKind: "cloudflare_tunnel" as const,
+        connectorToken: "token",
+        tunnelId: "tunnel-1",
+      };
+
+      yield* runtime.requestRecovery(config);
+
+      expect(Option.getOrNull(yield* Stream.runHead(runtime.recoveryRequests))).toEqual(config);
+    }),
+  );
+
+  it.effect("recovers a rejected tunnel without waiting for the connector to exit", () =>
+    Effect.gen(function* () {
+      const output = yield* Queue.unbounded<Uint8Array>();
+      const firstBatchObserved = yield* Deferred.make<void>();
+      const secondBatchObserved = yield* Deferred.make<void>();
+      const recoveryRequested = yield* Deferred.make<RelayManagedEndpointRuntimeConfig>();
+      const recoveryRetried = yield* Deferred.make<RelayManagedEndpointRuntimeConfig>();
+      let recoveryRequestCount = 0;
+      const spawned: Array<number> = [];
+      const encoder = new TextEncoder();
+      const connectorOutput = Stream.fromQueue(output).pipe(
+        Stream.tap((chunk) => {
+          const line = new TextDecoder().decode(chunk);
+          if (line === "first checkpoint\n") {
+            return Deferred.succeed(firstBatchObserved, undefined).pipe(Effect.asVoid);
+          }
+          if (line === "second checkpoint\n") {
+            return Deferred.succeed(secondBatchObserved, undefined).pipe(Effect.asVoid);
+          }
+          return Effect.void;
+        }),
+      );
+      const spawner = ChildProcessSpawner.make(() =>
+        Effect.gen(function* () {
+          const pid = 600;
+          spawned.push(pid);
+          const handle = makeHandle({ pid, onKill: () => {}, output: connectorOutput });
+          yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
+          return handle;
+        }),
+      );
+      const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
+      const config = {
+        providerKind: "cloudflare_tunnel" as const,
+        connectorToken: "token",
+        tunnelId: "deleted-tunnel",
+      };
+      const rejectedLine =
+        '2026-06-17T02:00:00Z ERR Register tunnel error from server side error="Unauthorized: Failed to get tunnel" connIndex=0\n';
+
+      yield* runtime.recoveryRequests.pipe(
+        Stream.runForEach((requested) => {
+          recoveryRequestCount += 1;
+          return Deferred.succeed(
+            recoveryRequestCount === 1 ? recoveryRequested : recoveryRetried,
+            requested,
+          ).pipe(Effect.asVoid);
+        }),
+        Effect.forkChild,
+      );
+      yield* runtime.applyConfig(config);
+
+      yield* Queue.offer(output, encoder.encode(rejectedLine.repeat(3)));
+      yield* Queue.offer(output, encoder.encode("first checkpoint\n"));
+      yield* Deferred.await(firstBatchObserved);
+      expect(yield* Deferred.isDone(recoveryRequested)).toBe(false);
+
+      yield* Queue.offer(
+        output,
+        encoder.encode(
+          "2026-06-17T02:00:00Z INF Registered tunnel connection connIndex=0\n" +
+            rejectedLine.repeat(3),
+        ),
+      );
+      yield* Queue.offer(output, encoder.encode("second checkpoint\n"));
+      yield* Deferred.await(secondBatchObserved);
+      expect(yield* Deferred.isDone(recoveryRequested)).toBe(false);
+
+      yield* Queue.offer(output, encoder.encode(rejectedLine));
+
+      expect(yield* Deferred.await(recoveryRequested)).toEqual(config);
+
+      yield* Queue.offer(output, encoder.encode(rejectedLine.repeat(4)));
+
+      expect(yield* Deferred.await(recoveryRetried)).toEqual(config);
+      expect(spawned).toEqual([600]);
+    }),
+  );
+
   it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () =>
     Effect.gen(function* () {
       const spawned: Array<ChildProcess.StandardCommand> = [];
@@ -388,6 +567,7 @@ describe("CloudManagedEndpointRuntime", () => {
       expect(status).toEqual({
         status: "failed",
         providerKind: "cloudflare_tunnel",
+        failure: "not-installed",
         reason: "The relay client is not installed.",
       });
       expect(spawn).not.toHaveBeenCalled();

apps/server/src/cloud/ManagedEndpointRuntime.ts

@@ -5,6 +5,7 @@ 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";
@@ -36,6 +37,7 @@ export type CloudManagedEndpointRuntimeStatus =
   | {
       readonly status: "failed";
       readonly providerKind: RelayManagedEndpointRuntimeConfig["providerKind"];
+      readonly failure: "unsupported-platform" | "not-installed" | "spawn-failed";
       readonly reason: string;
       readonly tunnelId?: string;
       readonly tunnelName?: string;
@@ -58,6 +60,9 @@ export class CloudManagedEndpointRuntime extends Context.Service<
     readonly applyConfig: (
       config: RelayManagedEndpointRuntimeConfig | null,
     ) => Effect.Effect<CloudManagedEndpointRuntimeStatus>;
+    readonly recoveryRequests: Stream.Stream<RelayManagedEndpointRuntimeConfig>;
+    readonly requestRecovery: (config: RelayManagedEndpointRuntimeConfig) => Effect.Effect<void>;
+    readonly withLinkStateLock: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
   }
 >()("t3/cloud/ManagedEndpointRuntime/CloudManagedEndpointRuntime") {}
 
@@ -68,6 +73,9 @@ interface ActiveConnector {
   readonly config: RelayManagedEndpointRuntimeConfig;
 }
 
+// Newly created tunnels can fail authorization briefly while Cloudflare propagates their token.
+const TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY = 4;
+
 export function classifyRelayClientOutput(line: string): "connected" | "warning" | "debug" {
   if (/\bRegistered tunnel connection\b/iu.test(line)) {
     return "connected";
@@ -78,6 +86,26 @@ export function classifyRelayClientOutput(line: string): "connected" | "warning"
   return /\b(?:ERR|WRN|FTL|PNC)\b/u.test(line) ? "warning" : "debug";
 }
 
+export function isRejectedRelayClientTunnelOutput(line: string): boolean {
+  return (
+    /\bRegister tunnel error from server side\b/iu.test(line) &&
+    /\bUnauthorized:\s*(?:Failed to get tunnel|Record for tunnel not found|Invalid tunnel secret)\b/iu.test(
+      line,
+    )
+  );
+}
+
+/** Connector startup failures can clear after installation or a later spawn attempt. */
+export function isRetryableManagedEndpointRuntimeStatus(status: unknown): boolean {
+  if (typeof status !== "object" || status === null || !("status" in status)) {
+    return false;
+  }
+  if (status.status !== "failed" || !("failure" in status)) {
+    return false;
+  }
+  return status.failure === "not-installed" || status.failure === "spawn-failed";
+}
+
 function runtimeConfigKey(config: RelayManagedEndpointRuntimeConfig): string {
   return JSON.stringify({
     providerKind: config.providerKind,
@@ -104,7 +132,9 @@ export const make = Effect.gen(function* () {
   const relayClient = yield* RelayClient.RelayClient;
   const activeRef = yield* Ref.make<ActiveConnector | null>(null);
   const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
+  const recoveryRequests = yield* Queue.sliding<RelayManagedEndpointRuntimeConfig>(1);
   const reconcileSemaphore = yield* Semaphore.make(1);
+  const linkStateSemaphore = yield* Semaphore.make(1);
   let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
 
   const stopActive = Effect.gen(function* () {
@@ -144,15 +174,18 @@ export const make = Effect.gen(function* () {
             tunnelId: connector.config.tunnelId,
             tunnelName: connector.config.tunnelName,
           });
+          yield* Queue.offer(recoveryRequests, connector.config);
           yield* reconcileConfig(desiredConfig);
         }),
       );
     }).pipe(
       Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })),
     );
 
-  const observeConnectorOutput = (connector: ActiveConnector) =>
-    connector.child.all.pipe(
+  const observeConnectorOutput = (connector: ActiveConnector) => {
+    let rejectedRegistrations = 0;
+
+    return connector.child.all.pipe(
       Stream.decodeText(),
       Stream.splitLines,
       Stream.map((line) => line.trim()),
@@ -167,8 +200,22 @@ export const make = Effect.gen(function* () {
         };
         switch (classifyRelayClientOutput(line)) {
           case "connected":
+            rejectedRegistrations = 0;
             return Effect.logInfo("Relay client tunnel connection registered", attributes);
           case "warning":
+            if (isRejectedRelayClientTunnelOutput(line)) {
+              rejectedRegistrations += 1;
+              if (rejectedRegistrations >= TUNNEL_AUTHORIZATION_FAILURES_BEFORE_RECOVERY) {
+                rejectedRegistrations = 0;
+                return Effect.logWarning(
+                  "Relay client tunnel was rejected; requesting recovery",
+                  attributes,
+                ).pipe(
+                  Effect.andThen(Queue.offer(recoveryRequests, connector.config)),
+                  Effect.asVoid,
+                );
+              }
+            }
             return Effect.logWarning("Relay client reported a transport warning", attributes);
           case "debug":
             return Effect.logDebug("Relay client output", attributes);
@@ -183,6 +230,7 @@ export const make = Effect.gen(function* () {
         }),
       ),
     );
+  };
 
   reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) {
     if (!config || config.providerKind !== "cloudflare_tunnel") {
@@ -214,6 +262,7 @@ export const make = Effect.gen(function* () {
       return {
         status: "failed",
         providerKind: "cloudflare_tunnel",
+        failure: executable.status === "unsupported" ? "unsupported-platform" : "not-installed",
         reason:
           executable.status === "unsupported"
             ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.`
@@ -256,6 +305,7 @@ export const make = Effect.gen(function* () {
             Effect.as({
               status: "failed",
               providerKind: "cloudflare_tunnel",
+              failure: "spawn-failed",
               reason: String(cause),
               ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
               ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
@@ -290,6 +340,7 @@ export const make = Effect.gen(function* () {
     return {
       status: "failed",
       providerKind: "cloudflare_tunnel",
+      failure: "spawn-failed",
       reason: "Relay client did not start.",
       ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
       ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
@@ -305,6 +356,9 @@ export const make = Effect.gen(function* () {
 
   const runtime = CloudManagedEndpointRuntime.of({
     applyConfig,
+    recoveryRequests: Stream.fromQueue(recoveryRequests),
+    requestRecovery: (config) => Queue.offer(recoveryRequests, config).pipe(Effect.asVoid),
+    withLinkStateLock: linkStateSemaphore.withPermits(1),
   });
 
   const initialConfig = yield* readRuntimeConfig.pipe(