| author |  Theo Browne <me@t3.gg> 2026-10-09 01:59:50-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-09 01:59:50-07:00 |
| log | 101f8b2f55df50d4f8d70ed8e6b8e0edf71a2f31 |
| tree | c2c6db526ac241e49d0a0b5d0117ea3b3196ef61 |
| parent | 43f8a8de17a7ac1baa7a3cf36d681856de2d8add |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): relay client updates no longer drop the host off T3 Connect (#17366)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>3 files changed, 123 insertions(+), 23 deletions(-)
apps
server
docs
internals
apps/server/src/cloud/ManagedEndpointRuntime.test.ts+82-6
| ... | @@ -851,7 +851,7 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -851,7 +851,7 @@ describe("CloudManagedEndpointRuntime", () => { |
| 851 | ); | 851 | ); |
| 852 | 852 | ||
| 853 | it.effect( | 853 | it.effect( |
| 854 | "runs an older relay client while the pinned one installs, then restarts on it and prunes", | 854 | "runs an older relay client while the pinned one installs, then hands off to it and prunes", |
| 855 | () => | 855 | () => |
| 856 | Effect.gen(function* () { | 856 | Effect.gen(function* () { |
| 857 | const olderClient = { | 857 | const olderClient = { |
| ... | @@ -870,7 +870,7 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -870,7 +870,7 @@ describe("CloudManagedEndpointRuntime", () => { |
| 870 | const releaseInstall = yield* Deferred.make<void>(); | 870 | const releaseInstall = yield* Deferred.make<void>(); |
| 871 | const pruned = yield* Deferred.make<void>(); | 871 | const pruned = yield* Deferred.make<void>(); |
| 872 | const spawned = yield* Queue.unbounded<string>(); | 872 | const spawned = yield* Queue.unbounded<string>(); |
| 873 | const killed: Array<string> = []; | 873 | const olderStopped = yield* Deferred.make<void>(); |
| 874 | const output = yield* Queue.unbounded<Uint8Array>(); | 874 | const output = yield* Queue.unbounded<Uint8Array>(); |
| 875 | const spawner = ChildProcessSpawner.make((command) => | 875 | const spawner = ChildProcessSpawner.make((command) => |
| 876 | Effect.gen(function* () { | 876 | Effect.gen(function* () { |
| ... | @@ -878,7 +878,11 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -878,7 +878,11 @@ describe("CloudManagedEndpointRuntime", () => { |
| 878 | yield* Queue.offer(spawned, executable); | 878 | yield* Queue.offer(spawned, executable); |
| 879 | const handle = makeHandle({ | 879 | const handle = makeHandle({ |
| 880 | pid: executable === pinnedClient.executablePath ? 2 : 1, | 880 | pid: executable === pinnedClient.executablePath ? 2 : 1, |
| 881 | onKill: () => killed.push(executable), | 881 | onKill: () => { |
| 882 | if (executable === olderClient.executablePath) { | ||
| 883 | Deferred.doneUnsafe(olderStopped, Effect.void); | ||
| 884 | } | ||
| 885 | }, | ||
| 882 | ...(executable === pinnedClient.executablePath | 886 | ...(executable === pinnedClient.executablePath |
| 883 | ? { output: Stream.fromQueue(output) } | 887 | ? { output: Stream.fromQueue(output) } |
| 884 | : {}), | 888 | : {}), |
| ... | @@ -916,7 +920,8 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -916,7 +920,8 @@ describe("CloudManagedEndpointRuntime", () => { |
| 916 | 920 | ||
| 917 | yield* Deferred.succeed(releaseInstall, undefined); | 921 | yield* Deferred.succeed(releaseInstall, undefined); |
| 918 | expect(yield* Queue.take(spawned)).toBe(pinnedClient.executablePath); | 922 | expect(yield* Queue.take(spawned)).toBe(pinnedClient.executablePath); |
| 919 | expect(killed).toEqual([olderClient.executablePath]); | 923 | // The older connector keeps serving until the pinned one registers. |
| 924 | expect(Option.isNone(yield* Deferred.poll(olderStopped))).toBe(true); | ||
| 920 | 925 | ||
| 921 | yield* Queue.offer( | 926 | yield* Queue.offer( |
| 922 | output, | 927 | output, |
| ... | @@ -924,6 +929,7 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -924,6 +929,7 @@ describe("CloudManagedEndpointRuntime", () => { |
| 924 | "2026-10-08T00:00:00Z INF Registered tunnel connection connIndex=0\n", | 929 | "2026-10-08T00:00:00Z INF Registered tunnel connection connIndex=0\n", |
| 925 | ), | 930 | ), |
| 926 | ); | 931 | ); |
| 932 | yield* Deferred.await(olderStopped); | ||
| 927 | yield* Deferred.await(pruned); | 933 | yield* Deferred.await(pruned); |
| 928 | }), | 934 | }), |
| 929 | ); | 935 | ); |
| ... | @@ -1118,7 +1124,7 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -1118,7 +1124,7 @@ describe("CloudManagedEndpointRuntime", () => { |
| 1118 | }), | 1124 | }), |
| 1119 | ); | 1125 | ); |
| 1120 | 1126 | ||
| 1121 | it.effect("requests recovery when the updated relay client fails to start", () => | 1127 | it.effect("keeps the older connector when the updated relay client fails to start", () => |
| 1122 | Effect.gen(function* () { | 1128 | Effect.gen(function* () { |
| 1123 | const olderClient = { | 1129 | const olderClient = { |
| 1124 | status: "available", | 1130 | status: "available", |
| ... | @@ -1133,18 +1139,26 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -1133,18 +1139,26 @@ describe("CloudManagedEndpointRuntime", () => { |
| 1133 | version: RelayClient.CLOUDFLARED_VERSION, | 1139 | version: RelayClient.CLOUDFLARED_VERSION, |
| 1134 | } as const; | 1140 | } as const; |
| 1135 | let installed = false; | 1141 | let installed = false; |
| 1142 | let olderKilled = false; | ||
| 1136 | const releaseInstall = yield* Deferred.make<void>(); | 1143 | const releaseInstall = yield* Deferred.make<void>(); |
| 1144 | const pinnedAttempted = yield* Deferred.make<void>(); | ||
| 1137 | const spawner = ChildProcessSpawner.make((command) => | 1145 | const spawner = ChildProcessSpawner.make((command) => |
| 1138 | Effect.gen(function* () { | 1146 | Effect.gen(function* () { |
| 1139 | const executable = ChildProcess.isStandardCommand(command) ? command.command : ""; | 1147 | const executable = ChildProcess.isStandardCommand(command) ? command.command : ""; |
| 1140 | if (executable === pinnedClient.executablePath) { | 1148 | if (executable === pinnedClient.executablePath) { |
| 1149 | yield* Deferred.succeed(pinnedAttempted, undefined); | ||
| 1141 | return yield* PlatformError.systemError({ | 1150 | return yield* PlatformError.systemError({ |
| 1142 | _tag: "PermissionDenied", | 1151 | _tag: "PermissionDenied", |
| 1143 | module: "ChildProcess", | 1152 | module: "ChildProcess", |
| 1144 | method: "spawn", | 1153 | method: "spawn", |
| 1145 | }); | 1154 | }); |
| 1146 | } | 1155 | } |
| 1147 | const handle = makeHandle({ pid: 6, onKill: () => {} }); | 1156 | const handle = makeHandle({ |
| 1157 | pid: 6, | ||
| 1158 | onKill: () => { | ||
| 1159 | olderKilled = true; | ||
| 1160 | }, | ||
| 1161 | }); | ||
| 1148 | yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); | 1162 | yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); |
| 1149 | return handle; | 1163 | return handle; |
| 1150 | }), | 1164 | }), |
| ... | @@ -1171,6 +1185,68 @@ describe("CloudManagedEndpointRuntime", () => { | ... | @@ -1171,6 +1185,68 @@ describe("CloudManagedEndpointRuntime", () => { |
| 1171 | const config = { providerKind: "cloudflare_tunnel", connectorToken: "token" } as const; | 1185 | const config = { providerKind: "cloudflare_tunnel", connectorToken: "token" } as const; |
| 1172 | expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 6 }); | 1186 | expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 6 }); |
| 1173 | 1187 | ||
| 1188 | yield* Deferred.succeed(releaseInstall, undefined); | ||
| 1189 | yield* Deferred.await(pinnedAttempted); | ||
| 1190 | // The failed swap leaves the older connector active and running. | ||
| 1191 | expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 6 }); | ||
| 1192 | expect(olderKilled).toBe(false); | ||
| 1193 | }), | ||
| 1194 | ); | ||
| 1195 | |||
| 1196 | it.effect("requests recovery when the older connector exited during a failed swap", () => | ||
| 1197 | Effect.gen(function* () { | ||
| 1198 | const olderClient = { | ||
| 1199 | status: "available", | ||
| 1200 | executablePath: "/managed/2025.9.0/cloudflared", | ||
| 1201 | source: "managed", | ||
| 1202 | version: "2025.9.0", | ||
| 1203 | } as const; | ||
| 1204 | const pinnedClient = { | ||
| 1205 | status: "available", | ||
| 1206 | executablePath: "/managed/pinned/cloudflared", | ||
| 1207 | source: "managed", | ||
| 1208 | version: RelayClient.CLOUDFLARED_VERSION, | ||
| 1209 | } as const; | ||
| 1210 | let installed = false; | ||
| 1211 | const releaseInstall = yield* Deferred.make<void>(); | ||
| 1212 | const spawner = ChildProcessSpawner.make((command) => | ||
| 1213 | Effect.gen(function* () { | ||
| 1214 | const executable = ChildProcess.isStandardCommand(command) ? command.command : ""; | ||
| 1215 | if (executable === pinnedClient.executablePath) { | ||
| 1216 | return yield* PlatformError.systemError({ | ||
| 1217 | _tag: "PermissionDenied", | ||
| 1218 | module: "ChildProcess", | ||
| 1219 | method: "spawn", | ||
| 1220 | }); | ||
| 1221 | } | ||
| 1222 | // The older connector has exited by the time the swap fails. | ||
| 1223 | const handle = makeHandle({ pid: 12, onKill: () => {}, isRunning: () => !installed }); | ||
| 1224 | yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); | ||
| 1225 | return handle; | ||
| 1226 | }), | ||
| 1227 | ); | ||
| 1228 | const runtime = yield* buildCloudManagedEndpointRuntime( | ||
| 1229 | spawner, | ||
| 1230 | Layer.succeed( | ||
| 1231 | RelayClient.RelayClient, | ||
| 1232 | RelayClient.RelayClient.of({ | ||
| 1233 | resolve: Effect.sync(() => (installed ? pinnedClient : olderClient)), | ||
| 1234 | install: Deferred.await(releaseInstall).pipe( | ||
| 1235 | Effect.andThen( | ||
| 1236 | Effect.sync(() => { | ||
| 1237 | installed = true; | ||
| 1238 | return pinnedClient; | ||
| 1239 | }), | ||
| 1240 | ), | ||
| 1241 | ), | ||
| 1242 | installWithProgress: () => Effect.die("unused"), | ||
| 1243 | pruneManagedVersions: Effect.void, | ||
| 1244 | }), | ||
| 1245 | ), | ||
| 1246 | ); | ||
| 1247 | const config = { providerKind: "cloudflare_tunnel", connectorToken: "token" } as const; | ||
| 1248 | expect(yield* runtime.applyConfig(config)).toMatchObject({ status: "running", pid: 12 }); | ||
| 1249 | |||
| 1174 | yield* Deferred.succeed(releaseInstall, undefined); | 1250 | yield* Deferred.succeed(releaseInstall, undefined); |
| 1175 | expect(Option.getOrThrow(yield* Stream.runHead(runtime.recoveryRequests))).toEqual(config); | 1251 | expect(Option.getOrThrow(yield* Stream.runHead(runtime.recoveryRequests))).toEqual(config); |
| 1176 | }), | 1252 | }), |
apps/server/src/cloud/ManagedEndpointRuntime.ts+36-13
| ... | @@ -2,6 +2,7 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay | ... | @@ -2,6 +2,7 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay |
| 2 | import * as RelayClient from "@t3tools/shared/relayClient"; | 2 | import * as RelayClient from "@t3tools/shared/relayClient"; |
| 3 | import * as Clock from "effect/Clock"; | 3 | import * as Clock from "effect/Clock"; |
| 4 | import * as Context from "effect/Context"; | 4 | import * as Context from "effect/Context"; |
| 5 | import * as Deferred from "effect/Deferred"; | ||
| 5 | import * as Duration from "effect/Duration"; | 6 | import * as Duration from "effect/Duration"; |
| 6 | import * as Effect from "effect/Effect"; | 7 | import * as Effect from "effect/Effect"; |
| 7 | import * as Exit from "effect/Exit"; | 8 | import * as Exit from "effect/Exit"; |
| ... | @@ -60,7 +61,8 @@ interface ActiveConnector { | ... | @@ -60,7 +61,8 @@ interface ActiveConnector { |
| 60 | readonly configKey: string; | 61 | readonly configKey: string; |
| 61 | readonly config: RelayManagedEndpointRuntimeConfig; | 62 | readonly config: RelayManagedEndpointRuntimeConfig; |
| 62 | readonly startedAtMillis: number; | 63 | readonly startedAtMillis: number; |
| 63 | readonly connected: Ref.Ref<boolean>; | 64 | /** Completes when the connector first registers a tunnel connection. */ |
| 65 | readonly registered: Deferred.Deferred<void>; | ||
| 64 | } | 66 | } |
| 65 | 67 | ||
| 66 | // A connector that exits before running this long is treated as part of a | 68 | // A connector that exits before running this long is treated as part of a |
| ... | @@ -252,9 +254,10 @@ export const make = Effect.gen(function* () { | ... | @@ -252,9 +254,10 @@ export const make = Effect.gen(function* () { |
| 252 | Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })), | 254 | Effect.catchCause((cause) => Effect.logWarning("Relay client supervisor failed", { cause })), |
| 253 | ); | 255 | ); |
| 254 | 256 | ||
| 255 | // Installs the pinned relay client in the background, then restarts the | 257 | // Installs the pinned relay client in the background, then moves the connector |
| 256 | // connector on it. The running connector, if any, keeps serving until the new | 258 | // to it. The new connector starts next to the old one on the same tunnel, and |
| 257 | // binary is installed and validated, so an update costs one brief reconnect. | 259 | // the old one stops only once the new one registers, so clients never see the |
| 260 | // relay lose this host during the swap. | ||
| 258 | const installPinnedRelayClient = Effect.gen(function* () { | 261 | const installPinnedRelayClient = Effect.gen(function* () { |
| 259 | if (yield* Ref.getAndSet(installInFlightRef, true)) return; | 262 | if (yield* Ref.getAndSet(installInFlightRef, true)) return; |
| 260 | yield* Effect.logInfo("Installing the pinned relay client", { | 263 | yield* Effect.logInfo("Installing the pinned relay client", { |
| ... | @@ -286,17 +289,37 @@ export const make = Effect.gen(function* () { | ... | @@ -286,17 +289,37 @@ export const make = Effect.gen(function* () { |
| 286 | ) { | 289 | ) { |
| 287 | return; | 290 | return; |
| 288 | } | 291 | } |
| 289 | yield* Effect.logInfo("Relay client installed; restarting the connector on it", { | 292 | yield* Effect.logInfo("Relay client installed; moving the connector to it", { |
| 290 | version: installed.version, | 293 | version: installed.version, |
| 291 | previousVersion: active?.executable.version, | 294 | previousVersion: active?.executable.version, |
| 292 | }); | 295 | }); |
| 293 | yield* stopActive; | 296 | // Detach the old connector so the reconcile starts a new one beside it. |
| 297 | const previous = yield* Ref.getAndSet(activeRef, null); | ||
| 294 | const status = yield* reconcileConfig(desiredConfig); | 298 | const status = yield* reconcileConfig(desiredConfig); |
| 295 | // The old connector is gone and no supervisor watches a failed spawn, | 299 | const next = yield* Ref.get(activeRef); |
| 296 | // so ask for recovery like an exited connector would. | 300 | if (!next) { |
| 297 | if (status.status === "failed") { | ||
| 298 | yield* Effect.logWarning("Relay client did not start after the update", status); | 301 | yield* Effect.logWarning("Relay client did not start after the update", status); |
| 299 | yield* Queue.offer(recoveryRequests, desiredConfig); | 302 | // A still-running old connector keeps serving and the retry loop |
| 303 | // tries the pin again. One that exited while detached was skipped | ||
| 304 | // by its supervisor, so ask for recovery like an exited connector. | ||
| 305 | const previousRunning = previous | ||
| 306 | ? yield* previous.child.isRunning.pipe(Effect.orElseSucceed(() => false)) | ||
| 307 | : false; | ||
| 308 | if (previous && previousRunning) { | ||
| 309 | yield* Ref.set(activeRef, previous); | ||
| 310 | } else { | ||
| 311 | yield* stopConnector(previous); | ||
| 312 | yield* Queue.offer(recoveryRequests, desiredConfig); | ||
| 313 | } | ||
| 314 | return; | ||
| 315 | } | ||
| 316 | // Runs in the new connector's scope: the old one stops when the new | ||
| 317 | // one registers, or as soon as the new one is stopped or replaced. | ||
| 318 | if (previous) { | ||
| 319 | yield* Deferred.await(next.registered).pipe( | ||
| 320 | Effect.ensuring(stopConnector(previous)), | ||
| 321 | Effect.forkIn(next.scope), | ||
| 322 | ); | ||
| 300 | } | 323 | } |
| 301 | }), | 324 | }), |
| 302 | ), | 325 | ), |
| ... | @@ -357,7 +380,7 @@ export const make = Effect.gen(function* () { | ... | @@ -357,7 +380,7 @@ export const make = Effect.gen(function* () { |
| 357 | const watchConnectorRegistration = (connector: ActiveConnector) => | 380 | const watchConnectorRegistration = (connector: ActiveConnector) => |
| 358 | Effect.gen(function* () { | 381 | Effect.gen(function* () { |
| 359 | yield* Effect.sleep(CONNECTOR_REGISTRATION_TIMEOUT); | 382 | yield* Effect.sleep(CONNECTOR_REGISTRATION_TIMEOUT); |
| 360 | if (yield* Ref.get(connector.connected)) return true; | 383 | if (yield* Deferred.isDone(connector.registered)) return true; |
| 361 | yield* Effect.logWarning( | 384 | yield* Effect.logWarning( |
| 362 | "Relay client has not registered a tunnel connection; requesting recovery", | 385 | "Relay client has not registered a tunnel connection; requesting recovery", |
| 363 | { | 386 | { |
| ... | @@ -389,7 +412,7 @@ export const make = Effect.gen(function* () { | ... | @@ -389,7 +412,7 @@ export const make = Effect.gen(function* () { |
| 389 | switch (classifyRelayClientOutput(line)) { | 412 | switch (classifyRelayClientOutput(line)) { |
| 390 | case "connected": | 413 | case "connected": |
| 391 | rejectedRegistrations = 0; | 414 | rejectedRegistrations = 0; |
| 392 | return Ref.set(connector.connected, true).pipe( | 415 | return Deferred.succeed(connector.registered, undefined).pipe( |
| 393 | Effect.andThen( | 416 | Effect.andThen( |
| 394 | Effect.logInfo("Relay client tunnel connection registered", attributes), | 417 | Effect.logInfo("Relay client tunnel connection registered", attributes), |
| 395 | ), | 418 | ), |
| ... | @@ -539,7 +562,7 @@ export const make = Effect.gen(function* () { | ... | @@ -539,7 +562,7 @@ export const make = Effect.gen(function* () { |
| 539 | configKey: nextConfigKey, | 562 | configKey: nextConfigKey, |
| 540 | config, | 563 | config, |
| 541 | startedAtMillis: yield* Clock.currentTimeMillis, | 564 | startedAtMillis: yield* Clock.currentTimeMillis, |
| 542 | connected: yield* Ref.make(false), | 565 | registered: yield* Deferred.make<void>(), |
| 543 | } satisfies ActiveConnector; | 566 | } satisfies ActiveConnector; |
| 544 | yield* Ref.set(activeRef, connector); | 567 | yield* Ref.set(activeRef, connector); |
| 545 | yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); | 568 | yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope); |
docs/internals/t3-connect.md+5-4
| ... | @@ -136,10 +136,11 @@ The host runs `cloudflared` pinned by `CLOUDFLARED_VERSION` in | ... | @@ -136,10 +136,11 @@ The host runs `cloudflared` pinned by `CLOUDFLARED_VERSION` in |
| 136 | [`relayClient.ts`](../../packages/shared/src/relayClient.ts), and bumping it there | 136 | [`relayClient.ts`](../../packages/shared/src/relayClient.ts), and bumping it there |
| 137 | (version, URLs, and checksums) is the whole release step. A linked host that | 137 | (version, URLs, and checksums) is the whole release step. A linked host that |
| 138 | starts a server with a new pin keeps its connector up on the newest older managed | 138 | starts a server with a new pin keeps its connector up on the newest older managed |
| 139 | release, installs the pinned one in the background, then restarts only the | 139 | release, installs the pinned one in the background, then starts a second |
| 140 | connector child on it. If that restart fails to spawn, the host requests recovery | 140 | connector on it for the same tunnel. The old connector stops only after the new |
| 141 | as it would for an exited connector. A host with no relay client installs one | 141 | one registers, so the swap never takes the host off the relay. If the new one |
| 142 | first. A `cloudflared` on `PATH` or an explicit `T3CODE_CLOUDFLARED_PATH` is the | 142 | fails to spawn, the old one keeps serving and the install retries later. A host |
| 143 | with no relay client installs one first. A `cloudflared` on `PATH` or an explicit `T3CODE_CLOUDFLARED_PATH` is the | ||
| 143 | user's choice and is never replaced, since linking asks before downloading. | 144 | user's choice and is never replaced, since linking asks before downloading. |
| 144 | 145 | ||
| 145 | Once the pinned release registers a connection, managed releases older than the | 146 | Once the pinned release registers a connection, managed releases older than the |