From 7527d08c1d1d592790b4c09970399d95acf03e9c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:54:55 +0000 Subject: [PATCH 1/3] fix(server): stop the relay client a crashed server left running A server that died without running its finalizers (crash, kill -9, OOM, force quit) left cloudflared running, and the next server started a second connector beside it. Nothing recorded the connector between runs. The server now records each relay client it starts (pid, start time and command line, plus its own pid and start time) under the state directory and forgets it once the client has stopped. On startup, before any connector starts, it stops a recorded client whose server is gone, but only while the pid still has the recorded start time and command line, so a recycled pid is never signalled. Linux and macOS only, like the OpenCode server ledger. Fixes pingdotgg/t3code#21701 AI-assisted; verified with the tests and tools listed above. --- .../src/cloud/ManagedEndpointRuntime.test.ts | 2 + .../src/cloud/ManagedEndpointRuntime.ts | 9 + .../src/cloud/RelayClientLedger.test.ts | 128 +++++++++ apps/server/src/cloud/RelayClientLedger.ts | 258 ++++++++++++++++++ apps/server/src/server.ts | 9 + 5 files changed, 406 insertions(+) create mode 100644 apps/server/src/cloud/RelayClientLedger.test.ts create mode 100644 apps/server/src/cloud/RelayClientLedger.ts diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index ba6d4a635c28..2e04ac364b77 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))); diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index c018bbf6c2f1..b2a77869efc5 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,6 +827,7 @@ export const make = Effect.gen(function* () { } if (!("status" in child)) { + forgetRelayClient = yield* relayClientLedger.track(Number(child.pid)); const connector = { child, executable, diff --git a/apps/server/src/cloud/RelayClientLedger.test.ts b/apps/server/src/cloud/RelayClientLedger.test.ts new file mode 100644 index 000000000000..310c008fa1d1 --- /dev/null +++ b/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(); + 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([]); + }), + ), + ); + + 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([]); + }), + ), + ); +}); 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 293d6670bbf0..09d24d97cae9 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -170,6 +170,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, @@ -539,6 +540,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 }); + }), + ), + ), ), ); From 3f974f686e5a1238f691836a9eab7efa2c9b6822 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:23:42 +0000 Subject: [PATCH 2/3] fix(server): record a relay client without an interrupt in between Keep the ledger entry and its forget handle together, and drop the test that only mirrored the forget call. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cloud/ManagedEndpointRuntime.ts | 2 +- apps/server/src/cloud/RelayClientLedger.test.ts | 16 ---------------- 2 files changed, 1 insertion(+), 17 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.ts b/apps/server/src/cloud/ManagedEndpointRuntime.ts index b2a77869efc5..b9173c69d0b8 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -827,7 +827,7 @@ export const make = Effect.gen(function* () { } if (!("status" in child)) { - forgetRelayClient = yield* relayClientLedger.track(Number(child.pid)); + forgetRelayClient = yield* Effect.uninterruptible(relayClientLedger.track(Number(child.pid))); const connector = { child, executable, diff --git a/apps/server/src/cloud/RelayClientLedger.test.ts b/apps/server/src/cloud/RelayClientLedger.test.ts index 310c008fa1d1..deee9ddc46e6 100644 --- a/apps/server/src/cloud/RelayClientLedger.test.ts +++ b/apps/server/src/cloud/RelayClientLedger.test.ts @@ -109,20 +109,4 @@ describe.each(observedPlatforms)("RelayClientLedger observing as %s", (platform) }), ), ); - - 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([]); - }), - ), - ); }); From 99ff9427ff3a40b6c6a05e253bd25daf047caeb4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:46:15 +0000 Subject: [PATCH 3/3] fix(server): an interrupted relay client activation still hands the client over Masking only the ledger write let a pending interrupt land right after it, before activeRef owned the connector, so its scope and the client stayed open: disabling left the client running and a retry started a second one. Record the client and store the connector in one uninterruptible region. Co-Authored-By: Claude Opus 5.5 --- .../src/cloud/ManagedEndpointRuntime.test.ts | 40 +++++++++++++++++++ .../src/cloud/ManagedEndpointRuntime.ts | 32 +++++++++------ 2 files changed, 59 insertions(+), 13 deletions(-) diff --git a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts index 2e04ac364b77..54baa2d16c0b 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.test.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.test.ts @@ -432,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 b9173c69d0b8..778878a71151 100644 --- a/apps/server/src/cloud/ManagedEndpointRuntime.ts +++ b/apps/server/src/cloud/ManagedEndpointRuntime.ts @@ -827,19 +827,25 @@ export const make = Effect.gen(function* () { } if (!("status" in child)) { - forgetRelayClient = yield* Effect.uninterruptible(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); + // 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);