diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index ba6d4a635c28..54baa2d16c0b 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/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(); + const recorded = yield* Deferred.make(); + 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" }); diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index c018bbf6c2f1..778878a71151 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/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 = 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(), - metricsAddress: yield* Deferred.make(), - protocol: yield* Ref.make(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(), + metricsAddress: yield* Deferred.make(), + protocol: yield* Ref.make(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); diff --git a/apps/server/src/cloud/RelayClientLedger.test.ts b/apps/server/src/cloud/RelayClientLedger.test.ts new file mode 100644 index 000000000000..deee9ddc46e6 --- /dev/null +++ b/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(); + 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 = + hostPlatform === "linux" ? ["linux", "darwin"] : hostPlatform === "darwin" ? ["darwin"] : []; + +describe.each(observedPlatforms)("RelayClientLedger observing as %s", (platform) => { + const run = (effect: Effect.Effect) => + 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([]); + }), + ), + ); +}); diff --git a/apps/server/src/cloud/RelayClientLedger.ts b/apps/server/src/cloud/RelayClientLedger.ts new file mode 100644 index 000000000000..dfd95e2df1ba --- /dev/null +++ b/apps/server/src/cloud/RelayClientLedger.ts @@ -0,0 +1,258 @@ +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; + +const ProcessIdentity = Schema.Struct({ + pid: Schema.Int, + startTime: Schema.String, + command: Schema.optional(Schema.String), +}); +type ProcessIdentity = typeof ProcessIdentity.Type; + +const RelayClientEntry = Schema.Struct({ + version: Schema.Literal(1), + client: ProcessIdentity, + // A process may rewrite its own argv, so the owner is identified by pid and start time. + owner: ProcessIdentity, +}); +type RelayClientEntry = typeof RelayClientEntry.Type; +const RelayClientEntryJson = Schema.fromJsonString(RelayClientEntry); +const decodeEntry = Schema.decodeUnknownOption(RelayClientEntryJson); +const encodeEntry = Schema.encodeEffect(RelayClientEntryJson); + +/** Remembers running relay clients so a server can stop the ones a crashed server left behind. */ +export class RelayClientLedger extends Context.Service< + RelayClientLedger, + { + /** Records a started relay client; the returned effect forgets it once it has stopped. */ + readonly track: (pid: number) => Effect.Effect>; + } +>()("t3/cloud/RelayClientLedger") {} + +const ENTRY_DIRECTORY = "relay-clients"; +const ENTRY_FILE = /^\d+\.json$/; +const STOP_POLL_INTERVAL = "50 millis"; +const STOP_POLL_ATTEMPTS = 40; + +interface ObservedProcess { + readonly startTime: string; + readonly command: string; + readonly zombie: boolean; +} + +// ps lstart needs C locale and UTC for stable parsing. +const DARWIN_PS_LINE = /^\s*(\d+)\s+(\S+)\s+(\w{3} \w{3} +\d{1,2} \d{2}:\d{2}:\d{2} \d{4})\s+(.*)$/; + +const processExists = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch (cause) { + return (cause as NodeJS.ErrnoException | undefined)?.code !== "ESRCH"; + } +}; + +const signal = (pid: number, name: NodeJS.Signals) => { + try { + process.kill(pid, name); + } catch {} +}; + +export const make = Effect.fn("RelayClientLedger.make")(function* (input: { + readonly stateDir: string; + readonly ownerPid?: number; +}) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const platform = yield* HostProcess.Platform; + const directory = path.join(input.stateDir, ENTRY_DIRECTORY); + const recordable = platform === "linux" || platform === "darwin"; + + const observeLinux = Effect.fnUntraced(function* (pid: number) { + const stat = yield* fs.readFileString(`/proc/${pid}/stat`); + const fields = stat + .slice(stat.lastIndexOf(")") + 2) + .trim() + .split(/\s+/); + const startTime = fields[19]; + if (startTime === undefined) return undefined; + const cmdline = yield* fs.readFileString(`/proc/${pid}/cmdline`); + return { + startTime, + command: cmdline.replace(/\0$/, "").replaceAll("\0", " "), + zombie: fields[0] === "Z", + } satisfies ObservedProcess; + }); + + const observeDarwin = (pid: number) => + spawner + .string( + ChildProcess.make("/bin/ps", ["-ww", "-o", "pid=,stat=,lstart=,args=", "-p", String(pid)], { + env: { LC_ALL: "C", TZ: "UTC" }, + extendEnv: true, + stdin: "ignore", + stderr: "ignore", + }), + ) + .pipe( + Effect.timeout("5 seconds"), + Effect.map((output) => { + for (const line of output.split("\n")) { + const match = DARWIN_PS_LINE.exec(line); + if (match === null || Number(match[1]) !== pid) continue; + return { + zombie: match[2]!.startsWith("Z"), + startTime: match[3]!, + command: match[4]!.trimEnd(), + } satisfies ObservedProcess; + } + return undefined; + }), + ); + + const observe = (pid: number): Effect.Effect => { + if (platform === "linux") return observeLinux(pid).pipe(Effect.orElseSucceed(() => undefined)); + if (platform === "darwin") + return observeDarwin(pid).pipe(Effect.orElseSucceed(() => undefined)); + return Effect.undefined; + }; + + // "unknown" means the pid exists but could not be read, so nothing may be concluded about it. + const compare = (identity: ProcessIdentity) => + observe(identity.pid).pipe( + Effect.map((observed): "same" | "gone" | "unknown" => { + if (observed === undefined) return processExists(identity.pid) ? "unknown" : "gone"; + return !observed.zombie && + observed.startTime === identity.startTime && + (identity.command === undefined || observed.command === identity.command) + ? "same" + : "gone"; + }), + ); + + const identify = (pid: number) => + observe(pid).pipe( + Effect.map((observed) => + observed === undefined || observed.zombie + ? undefined + : { pid, startTime: observed.startTime, command: observed.command }, + ), + ); + + const ownerProcess = yield* identify(input.ownerPid ?? process.pid); + const owner = + ownerProcess === undefined + ? undefined + : { pid: ownerProcess.pid, startTime: ownerProcess.startTime }; + + const track: RelayClientLedger["Service"]["track"] = Effect.fn("RelayClientLedger.track")( + function* (pid) { + if (!recordable) return Effect.void; + const client = yield* identify(pid); + if (owner === undefined || client?.command === undefined) { + yield* Effect.logWarning( + "Could not record the relay client; it will keep running if this server crashes", + { pid }, + ); + return Effect.void; + } + const entryPath = path.join(directory, `${pid}.json`); + yield* writeFileStringAtomically({ + filePath: entryPath, + contents: yield* encodeEntry({ version: 1, client, owner }), + }).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); + return fs + .remove(entryPath, { force: true }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not forget a stopped relay client", { cause }), + ), + ); + }, + Effect.catchCause((cause) => + Effect.logWarning("Could not record the relay client", { cause }).pipe( + Effect.as(Effect.void), + ), + ), + ); + + // Signals only a process that still has the recorded pid, start time and command line, + // so a recycled pid is never touched. Returns whether the client is gone. + const stopOrphan = Effect.fn("RelayClientLedger.stopOrphan")(function* (client: ProcessIdentity) { + let state = yield* compare(client); + if (state !== "same") return state === "gone"; + yield* Effect.logInfo("Stopping a relay client left by a previous T3 Code server", { + pid: client.pid, + }); + for (const name of ["SIGTERM", "SIGKILL"] as const) { + signal(client.pid, name); + for (let attempt = 0; attempt < STOP_POLL_ATTEMPTS && state === "same"; attempt++) { + yield* Effect.sleep(STOP_POLL_INTERVAL); + state = yield* compare(client); + } + if (state !== "same") break; + } + return state === "gone"; + }); + + const reapEntry = Effect.fn("RelayClientLedger.reapEntry")( + function* (entryPath: string) { + const entry = decodeEntry(yield* fs.readFileString(entryPath)); + if (Option.isSome(entry)) { + if ((yield* compare(entry.value.owner)) !== "gone") return; + if (!(yield* stopOrphan(entry.value.client))) return; + } + yield* fs.remove(entryPath, { force: true }); + }, + (effect, entryPath) => + effect.pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not clean up a recorded relay client", { entryPath, cause }), + ), + ), + ); + + const reapOrphans = Effect.gen(function* () { + if (!recordable) return; + const names = yield* fs.readDirectory(directory).pipe(Effect.orElseSucceed(() => [])); + yield* Effect.forEach( + names.filter((name) => ENTRY_FILE.test(name)), + (name) => reapEntry(path.join(directory, name)), + { concurrency: "unbounded", discard: true }, + ); + }); + + return { track, reapOrphans }; +}); + +// Reaping finishes before any connector starts, so exactly one serves the tunnel. +export const layer = (input: { readonly stateDir: string }) => + Layer.effect( + RelayClientLedger, + Effect.gen(function* () { + const ledger = yield* make(input); + yield* ledger.reapOrphans; + return RelayClientLedger.of({ track: ledger.track }); + }), + ); + +export const layerTest = Layer.succeed( + RelayClientLedger, + RelayClientLedger.of({ + track: Effect.fn("RelayClientLedger.track")(function () { + return Effect.succeed(Effect.void); + }), + }), +); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index dae95787afe8..39c89f769be5 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -175,6 +175,7 @@ import * as ConnectReachability from "./cloud/ConnectReachability.ts"; import { pendingServiceUpdateExists } from "./cloud/updateHandoff.ts"; import * as RelayTracing from "./cloud/relayTracing.ts"; import * as CloudManagedEndpointRuntime from "./cloud/ManagedEndpointRuntime.ts"; +import * as RelayClientLedger from "./cloud/RelayClientLedger.ts"; import { MANAGED_TUNNEL_FIRST_REGISTRATION_JITTER, makeManagedTunnelRecovery, @@ -547,6 +548,14 @@ const layerCloudManagedEndpointRuntime = Layer.mergeAll( CloudManagedEndpointRuntime.layer.pipe( Layer.provide(ServerSecretStore.layer), Layer.provide(layerRelayClient), + Layer.provide( + Layer.unwrap( + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + return RelayClientLedger.layer({ stateDir: config.stateDir }); + }), + ), + ), ), );