Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 82 additions & 6 deletions apps/server/src/cloud/ManagedEndpointRuntime.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -851,7 +851,7 @@ describe("CloudManagedEndpointRuntime", () => {
);

it.effect(
"runs an older relay client while the pinned one installs, then restarts on it and prunes",
"runs an older relay client while the pinned one installs, then hands off to it and prunes",
() =>
Effect.gen(function* () {
const olderClient = {
Expand All @@ -870,15 +870,19 @@ describe("CloudManagedEndpointRuntime", () => {
const releaseInstall = yield* Deferred.make<void>();
const pruned = yield* Deferred.make<void>();
const spawned = yield* Queue.unbounded<string>();
const killed: Array<string> = [];
const olderStopped = yield* Deferred.make<void>();
const output = yield* Queue.unbounded<Uint8Array>();
const spawner = ChildProcessSpawner.make((command) =>
Effect.gen(function* () {
const executable = ChildProcess.isStandardCommand(command) ? command.command : "";
yield* Queue.offer(spawned, executable);
const handle = makeHandle({
pid: executable === pinnedClient.executablePath ? 2 : 1,
onKill: () => killed.push(executable),
onKill: () => {
if (executable === olderClient.executablePath) {
Deferred.doneUnsafe(olderStopped, Effect.void);
}
},
...(executable === pinnedClient.executablePath
? { output: Stream.fromQueue(output) }
: {}),
Expand Down Expand Up @@ -916,14 +920,16 @@ describe("CloudManagedEndpointRuntime", () => {

yield* Deferred.succeed(releaseInstall, undefined);
expect(yield* Queue.take(spawned)).toBe(pinnedClient.executablePath);
expect(killed).toEqual([olderClient.executablePath]);
// The older connector keeps serving until the pinned one registers.
expect(Option.isNone(yield* Deferred.poll(olderStopped))).toBe(true);

yield* Queue.offer(
output,
new TextEncoder().encode(
"2026-10-08T00:00:00Z INF Registered tunnel connection connIndex=0\n",
),
);
yield* Deferred.await(olderStopped);
yield* Deferred.await(pruned);
}),
);
Expand Down Expand Up @@ -1118,7 +1124,7 @@ describe("CloudManagedEndpointRuntime", () => {
}),
);

it.effect("requests recovery when the updated relay client fails to start", () =>
it.effect("keeps the older connector when the updated relay client fails to start", () =>
Effect.gen(function* () {
const olderClient = {
status: "available",
Expand All @@ -1133,18 +1139,26 @@ describe("CloudManagedEndpointRuntime", () => {
version: RelayClient.CLOUDFLARED_VERSION,
} as const;
let installed = false;
let olderKilled = false;
const releaseInstall = yield* Deferred.make<void>();
const pinnedAttempted = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make((command) =>
Effect.gen(function* () {
const executable = ChildProcess.isStandardCommand(command) ? command.command : "";
if (executable === pinnedClient.executablePath) {
yield* Deferred.succeed(pinnedAttempted, undefined);
return yield* PlatformError.systemError({
_tag: "PermissionDenied",
module: "ChildProcess",
method: "spawn",
});
}
const handle = makeHandle({ pid: 6, onKill: () => {} });
const handle = makeHandle({
pid: 6,
onKill: () => {
olderKilled = true;
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
return handle;
}),
Expand All @@ -1171,6 +1185,68 @@ describe("CloudManagedEndpointRuntime", () => {
const config = { providerKind: "cloudflare_tunnel", connectorToken: "token" } as const;
expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 6 });

yield* Deferred.succeed(releaseInstall, undefined);
yield* Deferred.await(pinnedAttempted);
// The failed swap leaves the older connector active and running.
expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 6 });
expect(olderKilled).toBe(false);
}),
);

it.effect("requests recovery when the older connector exited during a failed swap", () =>
Effect.gen(function* () {
const olderClient = {
status: "available",
executablePath: "/managed/2025.9.0/cloudflared",
source: "managed",
version: "2025.9.0",
} as const;
const pinnedClient = {
status: "available",
executablePath: "/managed/pinned/cloudflared",
source: "managed",
version: RelayClient.CLOUDFLARED_VERSION,
} as const;
let installed = false;
const releaseInstall = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make((command) =>
Effect.gen(function* () {
const executable = ChildProcess.isStandardCommand(command) ? command.command : "";
if (executable === pinnedClient.executablePath) {
return yield* PlatformError.systemError({
_tag: "PermissionDenied",
module: "ChildProcess",
method: "spawn",
});
}
// The older connector has exited by the time the swap fails.
const handle = makeHandle({ pid: 12, onKill: () => {}, isRunning: () => !installed });
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
return handle;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(
spawner,
Layer.succeed(
RelayClient.RelayClient,
RelayClient.RelayClient.of({
resolve: Effect.sync(() => (installed ? pinnedClient : olderClient)),
install: Deferred.await(releaseInstall).pipe(
Effect.andThen(
Effect.sync(() => {
installed = true;
return pinnedClient;
}),
),
),
installWithProgress: () => Effect.die("unused"),
pruneManagedVersions: Effect.void,
}),
),
);
const config = { providerKind: "cloudflare_tunnel", connectorToken: "token" } as const;
expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 12 });

yield* Deferred.succeed(releaseInstall, undefined);
expect(Option.getOrThrow(yield* Stream.runHead(runtime.recoveryRequests))).toEqual(config);
}),
Expand Down
49 changes: 36 additions & 13 deletions apps/server/src/cloud/ManagedEndpointRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay
import * as RelayClient from "@t3tools/shared/relayClient";
import * as Clock from "effect/Clock";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
Expand Down Expand Up @@ -60,7 +61,8 @@ interface ActiveConnector {
readonly configKey: string;
readonly config: RelayManagedEndpointRuntimeConfig;
readonly startedAtMillis: number;
readonly connected: Ref.Ref<boolean>;
/** Completes when the connector first registers a tunnel connection. */
readonly registered: Deferred.Deferred<void>;
}

// A connector that exits before running this long is treated as part of a
Expand Down Expand Up @@ -252,9 +254,10 @@ export const make = Effect.gen(function* () {
Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })),
);

// Installs the pinned relay client in the background, then restarts the
// connector on it. The running connector, if any, keeps serving until the new
// binary is installed and validated, so an update costs one brief reconnect.
// Installs the pinned relay client in the background, then moves the connector
// to it. The new connector starts next to the old one on the same tunnel, and
// the old one stops only once the new one registers, so clients never see the
// relay lose this host during the swap.
const installPinnedRelayClient = Effect.gen(function* () {
if (yield* Ref.getAndSet(installInFlightRef, true)) return;
yield* Effect.logInfo("Installing the pinned relay client", {
Expand Down Expand Up @@ -286,17 +289,37 @@ export const make = Effect.gen(function* () {
) {
return;
}
yield* Effect.logInfo("Relay client installed; restarting the connector on it", {
yield* Effect.logInfo("Relay client installed; moving the connector to it", {
version: installed.version,
previousVersion: active?.executable.version,
});
yield* stopActive;
// Detach the old connector so the reconcile starts a new one beside it.
const previous = yield* Ref.getAndSet(activeRef, null);
Comment thread
t3dotgg marked this conversation as resolved.
const status = yield* reconcileConfig(desiredConfig);
// The old connector is gone and no supervisor watches a failed spawn,
// so ask for recovery like an exited connector would.
if (status.status === "failed") {
const next = yield* Ref.get(activeRef);
if (!next) {
yield* Effect.logWarning("Relay client did not start after the update", status);
yield* Queue.offer(recoveryRequests, desiredConfig);
// A still-running old connector keeps serving and the retry loop
// tries the pin again. One that exited while detached was skipped
// by its supervisor, so ask for recovery like an exited connector.
const previousRunning = previous
? yield* previous.child.isRunning.pipe(Effect.orElseSucceed(() => false))
: false;
if (previous && previousRunning) {
yield* Ref.set(activeRef, previous);
} else {
yield* stopConnector(previous);
yield* Queue.offer(recoveryRequests, desiredConfig);
}
return;
}
// Runs in the new connector's scope: the old one stops when the new
// one registers, or as soon as the new one is stopped or replaced.
if (previous) {
yield* Deferred.await(next.registered).pipe(
Effect.ensuring(stopConnector(previous)),
Effect.forkIn(next.scope),
);
Comment thread
t3dotgg marked this conversation as resolved.
}
}),
),
Expand Down Expand Up @@ -357,7 +380,7 @@ export const make = Effect.gen(function* () {
const watchConnectorRegistration = (connector: ActiveConnector) =>
Effect.gen(function* () {
yield* Effect.sleep(CONNECTOR_REGISTRATION_TIMEOUT);
if (yield* Ref.get(connector.connected)) return true;
if (yield* Deferred.isDone(connector.registered)) return true;
yield* Effect.logWarning(
"Relay client has not registered a tunnel connection; requesting recovery",
{
Expand Down Expand Up @@ -389,7 +412,7 @@ export const make = Effect.gen(function* () {
switch (classifyRelayClientOutput(line)) {
case "connected":
rejectedRegistrations = 0;
return Ref.set(connector.connected, true).pipe(
return Deferred.succeed(connector.registered, undefined).pipe(
Effect.andThen(
Effect.logInfo("Relay client tunnel connection registered", attributes),
),
Expand Down Expand Up @@ -539,7 +562,7 @@ export const make = Effect.gen(function* () {
configKey: nextConfigKey,
config,
startedAtMillis: yield* Clock.currentTimeMillis,
connected: yield* Ref.make(false),
registered: yield* Deferred.make<void>(),
} satisfies ActiveConnector;
yield* Ref.set(activeRef, connector);
yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
Expand Down
9 changes: 5 additions & 4 deletions docs/internals/t3-connect.md
Original file line number Diff line number Diff line change
Expand Up @@ -136,10 +136,11 @@ The host runs `cloudflared` pinned by `CLOUDFLARED_VERSION` in
[`relayClient.ts`](../../packages/shared/src/relayClient.ts), and bumping it there
(version, URLs, and checksums) is the whole release step. A linked host that
starts a server with a new pin keeps its connector up on the newest older managed
release, installs the pinned one in the background, then restarts only the
connector child on it. If that restart fails to spawn, the host requests recovery
as it would for an exited connector. A host with no relay client installs one
first. A `cloudflared` on `PATH` or an explicit `T3CODE_CLOUDFLARED_PATH` is the
release, installs the pinned one in the background, then starts a second
connector on it for the same tunnel. The old connector stops only after the new
one registers, so the swap never takes the host off the relay. If the new one
fails to spawn, the old one keeps serving and the install retries later. A host
with no relay client installs one first. A `cloudflared` on `PATH` or an explicit `T3CODE_CLOUDFLARED_PATH` is the
user's choice and is never replaced, since linking asks before downloading.

Once the pinned release registers a connection, managed releases older than the
Expand Down
Loading