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)));
 

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,6 +827,7 @@ export const make = Effect.gen(function* () {
     }
 
     if (!("status" in child)) {
+      forgetRelayClient = yield* relayClientLedger.track(Number(child.pid));
       const connector = {
         child,
         executable,

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

@@ -0,0 +1,128 @@
+// @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([]);
+      }),
+    ),
+  );
+
+  it.live("forgets a client once it is stopped", () =>
+    run(
+      Effect.gen(function* () {
+        const fs = yield* FileSystem.FileSystem;
+        const stateDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-relay-ledger-" });
+        const client = yield* spawnSleep;
+        const ledger = yield* RelayClientLedger.make({ stateDir });
+        const forget = yield* ledger.track(client.pid);
+        expect(yield* entryNames(stateDir)).toEqual([`${client.pid}.json`]);
+        yield* client.kill;
+        yield* forget;
+        expect(yield* entryNames(stateDir)).toEqual([]);
+      }),
+    ),
+  );
+});