fix(server): stop the relay client a crashed server left running

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

@@ -30,6 +30,7 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 import * as ServerConfig from "../config.ts";
 import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";
 import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
+import * as RelayClientLedger from "./RelayClientLedger.ts";
 
 const layerRelayClientAvailable = Layer.succeed(
   RelayClient.RelayClient,
@@ -74,6 +75,7 @@ const runtimeDependencies = (
     Layer.succeed(HttpClient.HttpClient, httpClient),
     relayClientLayer,
     secretStoreLayer,
+    RelayClientLedger.layerTest,
     ServerConfig.layerTest(process.cwd(), { prefix: "t3-managed-endpoint-" }),
   ).pipe(Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)));
 
@@ -430,6 +432,46 @@ describe("CloudManagedEndpointRuntime", () => {
     }),
   );
 
+  it.effect("still owns a relay client when activation is interrupted while recording it", () =>
+    Effect.gen(function* () {
+      const fixture = yield* makeTransportSpawner();
+      const recording = yield* Deferred.make<void>();
+      const recorded = yield* Deferred.make<void>();
+      const ledger = Layer.succeed(
+        RelayClientLedger.RelayClientLedger,
+        RelayClientLedger.RelayClientLedger.of({
+          track: () =>
+            Deferred.succeed(recording, undefined).pipe(
+              Effect.andThen(Deferred.await(recorded)),
+              Effect.as(Effect.void),
+            ),
+        }),
+      );
+      const context = yield* Layer.build(
+        ManagedEndpointRuntime.layer.pipe(
+          Layer.provide(Layer.merge(runtimeDependencies(fixture.spawner), ledger)),
+        ),
+      );
+      const runtime = yield* Effect.service(
+        ManagedEndpointRuntime.CloudManagedEndpointRuntime,
+      ).pipe(Effect.provide(context));
+
+      const activation = yield* Effect.forkChild(runtime.applyConfig(tunnelConfig));
+      yield* Deferred.await(recording);
+      const interrupting = yield* Effect.forkChild(Fiber.interrupt(activation), {
+        startImmediately: true,
+      });
+      yield* Deferred.succeed(recorded, undefined);
+      yield* Fiber.join(interrupting);
+
+      yield* runtime.applyConfig(null);
+      expect(fixture.killed).toEqual([800]);
+    }).pipe(
+      Effect.provideService(HostProcess.Environment, {}),
+      Effect.provide(ConfigProvider.layer(ConfigProvider.fromEnv({ env: {} }))),
+    ),
+  );
+
   it.effect("preserves an explicit transport choice despite repeated QUIC stalls", () =>
     Effect.gen(function* () {
       const fixture = yield* makeTransportSpawner({ transport: "quic" });

apps/server/src/cloud/ManagedEndpointRuntime.ts

@@ -28,6 +28,7 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 import * as ServerConfig from "../config.ts";
 import * as ManagedLimits from "../policy/managedLimits.ts";
 import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
+import * as RelayClientLedger from "./RelayClientLedger.ts";
 
 export type CloudManagedEndpointRuntimeStatus =
   | {
@@ -210,6 +211,7 @@ const stopConnector = (connector: ActiveConnector | null) =>
 export const make = Effect.gen(function* () {
   const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
   const relayClient = yield* RelayClient.RelayClient;
+  const relayClientLedger = yield* RelayClientLedger.RelayClientLedger;
   const secrets = yield* ServerSecretStore.ServerSecretStore;
   const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
   // Without T3 Connect no connector ever runs, whoever asks for one.
@@ -755,6 +757,12 @@ export const make = Effect.gen(function* () {
 
     const http2 = yield* Ref.get(http2Ref);
     const connectorScope = yield* Scope.make("sequential");
+    // Added before the spawn so it runs after the spawner's finalizer has stopped the client.
+    let forgetRelayClient: Effect.Effect<void> = Effect.void;
+    yield* Scope.addFinalizer(
+      connectorScope,
+      Effect.suspend(() => forgetRelayClient),
+    );
     const child = yield* spawner
       .spawn(
         ChildProcess.make(
@@ -819,18 +827,25 @@ export const make = Effect.gen(function* () {
     }
 
     if (!("status" in child)) {
-      const connector = {
-        child,
-        executable,
-        scope: connectorScope,
-        configKey: nextConfigKey,
-        config,
-        startedAtMillis: yield* Clock.currentTimeMillis,
-        registered: yield* Deferred.make<void>(),
-        metricsAddress: yield* Deferred.make<string>(),
-        protocol: yield* Ref.make<string | null>(null),
-      } satisfies ActiveConnector;
-      yield* Ref.set(activeRef, connector);
+      // An interrupt before activeRef owns the connector would leave its scope, and the client, open.
+      const connector = yield* Effect.uninterruptible(
+        Effect.gen(function* () {
+          forgetRelayClient = yield* relayClientLedger.track(Number(child.pid));
+          const connector = {
+            child,
+            executable,
+            scope: connectorScope,
+            configKey: nextConfigKey,
+            config,
+            startedAtMillis: yield* Clock.currentTimeMillis,
+            registered: yield* Deferred.make<void>(),
+            metricsAddress: yield* Deferred.make<string>(),
+            protocol: yield* Ref.make<string | null>(null),
+          } satisfies ActiveConnector;
+          yield* Ref.set(activeRef, connector);
+          return connector;
+        }),
+      );
       yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
       yield* Effect.forkIn(superviseConnector(connector), connectorScope);
       yield* Effect.forkIn(watchConnectorRegistration(connector), connectorScope);

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

@@ -0,0 +1,112 @@
+// @effect-diagnostics nodeBuiltinImport:off
+import * as NodeChildProcess from "node:child_process";
+
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { describe, expect, it } from "@effect/vitest";
+import * as HostProcess from "@t3tools/shared/HostProcess";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Path from "effect/Path";
+
+import * as RelayClientLedger from "./RelayClientLedger.ts";
+
+const isAlive = (pid: number) => {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+};
+
+const spawnSleep = Effect.acquireRelease(
+  Effect.gen(function* () {
+    const child = NodeChildProcess.spawn("sleep", ["600"], { stdio: "ignore" });
+    const pid = child.pid;
+    if (pid === undefined) return yield* Effect.die("spawn failed");
+    const exited = yield* Deferred.make<void>();
+    child.once("exit", () => Deferred.doneUnsafe(exited, Effect.void));
+    return {
+      pid,
+      kill: Effect.sync(() => child.kill("SIGKILL")).pipe(Effect.andThen(Deferred.await(exited))),
+    };
+  }),
+  ({ kill }) => kill,
+);
+
+const entryNames = (stateDir: string) =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    return yield* fs
+      .readDirectory(path.join(stateDir, "relay-clients"))
+      .pipe(Effect.orElseSucceed(() => []));
+  });
+
+const hostPlatform = HostProcess.Platform.defaultValue();
+// procps accepts the same `ps` flags as macOS, so Linux also covers the macOS path.
+const observedPlatforms: ReadonlyArray<NodeJS.Platform> =
+  hostPlatform === "linux" ? ["linux", "darwin"] : hostPlatform === "darwin" ? ["darwin"] : [];
+
+describe.each(observedPlatforms)("RelayClientLedger observing as %s", (platform) => {
+  const run = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+    effect.pipe(
+      Effect.scoped,
+      Effect.provideService(HostProcess.Platform, platform),
+      Effect.provide(NodeServices.layer),
+    );
+
+  it.live("stops a client left by a dead server and keeps one whose server still runs", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const stateDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-relay-ledger-" });
+        const deadOwner = yield* spawnSleep;
+        const liveOwner = yield* spawnSleep;
+        const orphan = yield* spawnSleep;
+        const adopted = yield* spawnSleep;
+
+        const dead = yield* RelayClientLedger.make({ stateDir, ownerPid: deadOwner.pid });
+        yield* Effect.asVoid(dead.track(orphan.pid));
+        const live = yield* RelayClientLedger.make({ stateDir, ownerPid: liveOwner.pid });
+        yield* Effect.asVoid(live.track(adopted.pid));
+        yield* deadOwner.kill;
+
+        const restarted = yield* RelayClientLedger.make({ stateDir });
+        yield* restarted.reapOrphans;
+
+        expect(isAlive(orphan.pid)).toBe(false);
+        expect(isAlive(adopted.pid)).toBe(true);
+        expect(yield* entryNames(stateDir)).toEqual([`${adopted.pid}.json`]);
+      }),
+    ),
+  );
+
+  it.live("never signals a recycled pid and forgets the stale record", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const path = yield* Path.Path;
+        const stateDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-relay-ledger-" });
+        const deadOwner = yield* spawnSleep;
+        const unrelated = yield* spawnSleep;
+
+        const dead = yield* RelayClientLedger.make({ stateDir, ownerPid: deadOwner.pid });
+        yield* Effect.asVoid(dead.track(unrelated.pid));
+        yield* deadOwner.kill;
+        // The recorded client has exited and its pid now belongs to another process.
+        const entryPath = path.join(stateDir, "relay-clients", `${unrelated.pid}.json`);
+        const entry = JSON.parse(yield* fs.readFileString(entryPath));
+        entry.client.startTime = "0";
+        yield* fs.writeFileString(entryPath, JSON.stringify(entry));
+
+        const restarted = yield* RelayClientLedger.make({ stateDir });
+        yield* restarted.reapOrphans;
+
+        expect(isAlive(unrelated.pid)).toBe(true);
+        expect(yield* entryNames(stateDir)).toEqual([]);
+      }),
+    ),
+  );
+});