authorgravatar for 275381468+derektrimm@users.noreply.github.comDerek Trimm <275381468+derektrimm@users.noreply.github.com> 2026-09-03 20:16:16-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-03 18:16:16-07:00
logb90898077e60290a1eb7bea4224f325fd3bb1595
treebfb861465ea0e718a96c0bb192dbf41fa403e5a6
parent2b96220f00de09b61327c8e881b7f509f8cc5a79
signature Signed by PGP key B5690EEEBB952194

fix(server): back off relay client restarts after rapid exits (#8788)

2 files changed, 174 insertions(+), 1 deletions(-)

apps/server/src/cloud/ManagedEndpointRuntime.test.ts+122
...@@ -1,6 +1,7 @@...@@ -1,6 +1,7 @@
1import { describe, expect, it } from "@effect/vitest";1import { describe, expect, it } from "@effect/vitest";
2import { vi } from "vite-plus/test";2import { vi } from "vite-plus/test";
3import * as Deferred from "effect/Deferred";3import * as Deferred from "effect/Deferred";
4import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";6import * as Fiber from "effect/Fiber";
6import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
...@@ -8,6 +9,7 @@ import * as Option from "effect/Option";...@@ -8,6 +9,7 @@ import * as Option from "effect/Option";
8import * as PlatformError from "effect/PlatformError";9import * as PlatformError from "effect/PlatformError";
9import * as Sink from "effect/Sink";10import * as Sink from "effect/Sink";
10import * as Stream from "effect/Stream";11import * as Stream from "effect/Stream";
12import * as TestClock from "effect/testing/TestClock";
11import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";13import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
12import * as RelayClient from "@t3tools/shared/relayClient";14import * as RelayClient from "@t3tools/shared/relayClient";
1315
...@@ -283,6 +285,126 @@ describe("CloudManagedEndpointRuntime", () => {...@@ -283,6 +285,126 @@ describe("CloudManagedEndpointRuntime", () => {
283 }),285 }),
284 );286 );
285287
288 const makeCrashLoopSpawner = (basePid: number, slots: number) =>
289 Effect.gen(function* () {
290 const spawned: Array<number> = [];
291 const exits: Array<Deferred.Deferred<ChildProcessSpawner.ExitCode>> = [];
292 const spawnSignals: Array<Deferred.Deferred<void>> = [];
293 for (let index = 0; index < slots; index += 1) {
294 exits.push(yield* Deferred.make<ChildProcessSpawner.ExitCode>());
295 spawnSignals.push(yield* Deferred.make<void>());
296 }
297 const spawner = ChildProcessSpawner.make(() =>
298 Effect.gen(function* () {
299 const index = spawned.length;
300 spawned.push(basePid + index);
301 yield* Deferred.succeed(spawnSignals[index]!, undefined);
302 const handle = makeHandle({
303 pid: basePid + index,
304 exitCode: Deferred.await(exits[index]!),
305 onKill: () => {},
306 });
307 yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
308 return handle;
309 }),
310 );
311 return { spawner, spawned, exits, spawnSignals };
312 });
313
314 it.effect("backs off restarts while the connector crash-loops", () =>
315 Effect.gen(function* () {
316 const { spawner, spawned, exits, spawnSignals } = yield* makeCrashLoopSpawner(600, 4);
317 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
318
319 yield* runtime.applyConfig({
320 providerKind: "cloudflare_tunnel",
321 connectorToken: "token",
322 tunnelId: "tunnel-1",
323 });
324 expect(spawned).toEqual([600]);
325
326 // The first crash restarts immediately.
327 yield* Deferred.succeed(exits[0]!, ChildProcessSpawner.ExitCode(1));
328 yield* Deferred.await(spawnSignals[1]!);
329 expect(spawned).toEqual([600, 601]);
330
331 // The second rapid crash waits out the base delay before restarting.
332 yield* Deferred.succeed(exits[1]!, ChildProcessSpawner.ExitCode(1));
333 yield* TestClock.adjust(Duration.millis(999));
334 expect(spawned).toEqual([600, 601]);
335 yield* TestClock.adjust(Duration.millis(1));
336 yield* Deferred.await(spawnSignals[2]!);
337 expect(spawned).toEqual([600, 601, 602]);
338
339 // The third rapid crash doubles the delay.
340 yield* Deferred.succeed(exits[2]!, ChildProcessSpawner.ExitCode(1));
341 yield* TestClock.adjust(Duration.millis(1999));
342 expect(spawned).toEqual([600, 601, 602]);
343 yield* TestClock.adjust(Duration.millis(1));
344 yield* Deferred.await(spawnSignals[3]!);
345 expect(spawned).toEqual([600, 601, 602, 603]);
346 }).pipe(Effect.provide(TestClock.layer())),
347 );
348
349 it.effect("resets the backoff after the connector runs stably", () =>
350 Effect.gen(function* () {
351 const { spawner, spawned, exits, spawnSignals } = yield* makeCrashLoopSpawner(700, 5);
352 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
353
354 yield* runtime.applyConfig({
355 providerKind: "cloudflare_tunnel",
356 connectorToken: "token",
357 });
358
359 // One rapid crash arms the backoff.
360 yield* Deferred.succeed(exits[0]!, ChildProcessSpawner.ExitCode(1));
361 yield* Deferred.await(spawnSignals[1]!);
362
363 // The replacement stays up past the stable-uptime window, so its exit
364 // restarts immediately and the backoff starts over.
365 yield* TestClock.adjust(Duration.millis(30_000));
366 yield* Deferred.succeed(exits[1]!, ChildProcessSpawner.ExitCode(1));
367 yield* Deferred.await(spawnSignals[2]!);
368 yield* Deferred.succeed(exits[2]!, ChildProcessSpawner.ExitCode(1));
369 yield* Deferred.await(spawnSignals[3]!);
370
371 // The next rapid crash waits the base delay again, not a doubled one.
372 yield* Deferred.succeed(exits[3]!, ChildProcessSpawner.ExitCode(1));
373 yield* TestClock.adjust(Duration.millis(999));
374 expect(spawned).toEqual([700, 701, 702, 703]);
375 yield* TestClock.adjust(Duration.millis(1));
376 yield* Deferred.await(spawnSignals[4]!);
377 expect(spawned).toEqual([700, 701, 702, 703, 704]);
378 }).pipe(Effect.provide(TestClock.layer())),
379 );
380
381 it.effect("an explicit config change clears the backoff and preempts a delayed restart", () =>
382 Effect.gen(function* () {
383 const { spawner, spawned, exits, spawnSignals } = yield* makeCrashLoopSpawner(800, 3);
384 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
385
386 yield* runtime.applyConfig({
387 providerKind: "cloudflare_tunnel",
388 connectorToken: "token-1",
389 });
390 yield* Deferred.succeed(exits[0]!, ChildProcessSpawner.ExitCode(1));
391 yield* Deferred.await(spawnSignals[1]!);
392
393 // Leave the supervisor sleeping on the base delay, then change config.
394 yield* Deferred.succeed(exits[1]!, ChildProcessSpawner.ExitCode(1));
395 const status = yield* runtime.applyConfig({
396 providerKind: "cloudflare_tunnel",
397 connectorToken: "token-2",
398 });
399 expect(status).toMatchObject({ status: "running", pid: 802 });
400 expect(spawned).toEqual([800, 801, 802]);
401
402 // The preempted supervisor wakes later and must not spawn a duplicate.
403 yield* TestClock.adjust(Duration.millis(60_000));
404 expect(spawned).toEqual([800, 801, 802]);
405 }).pipe(Effect.provide(TestClock.layer())),
406 );
407
286 it.effect("serializes concurrent connector config changes", () =>408 it.effect("serializes concurrent connector config changes", () =>
287 Effect.gen(function* () {409 Effect.gen(function* () {
288 const spawned: Array<number> = [];410 const spawned: Array<number> = [];
apps/server/src/cloud/ManagedEndpointRuntime.ts+52-1
...@@ -1,6 +1,8 @@...@@ -1,6 +1,8 @@
1import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";1import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
2import * as RelayClient from "@t3tools/shared/relayClient";2import * as RelayClient from "@t3tools/shared/relayClient";
3import * as Clock from "effect/Clock";
3import * as Context from "effect/Context";4import * as Context from "effect/Context";
5import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";7import * as Exit from "effect/Exit";
6import * as Layer from "effect/Layer";8import * as Layer from "effect/Layer";
...@@ -66,8 +68,18 @@ interface ActiveConnector {...@@ -66,8 +68,18 @@ interface ActiveConnector {
66 readonly scope: Scope.Closeable;68 readonly scope: Scope.Closeable;
67 readonly configKey: string;69 readonly configKey: string;
68 readonly config: RelayManagedEndpointRuntimeConfig;70 readonly config: RelayManagedEndpointRuntimeConfig;
71 readonly startedAtMillis: number;
69}72}
7073
74// A connector that exits before running this long is treated as part of a
75// crash loop; one that stays up at least this long earns an immediate restart
76// again. Without the backoff below, a relay client that fails instantly (a
77// stale version-manager shim, a bad binary) respawns ~100 times per second
78// until the accumulated tracing exhausts the V8 heap.
79const RELAY_RESTART_STABLE_UPTIME_MS = 30_000;
80const RELAY_RESTART_BACKOFF_BASE_MS = 1_000;
81const RELAY_RESTART_BACKOFF_MAX_MS = 60_000;
82
71export function classifyRelayClientOutput(line: string): "connected" | "warning" | "debug" {83export function classifyRelayClientOutput(line: string): "connected" | "warning" | "debug" {
72 if (/\bRegistered tunnel connection\b/iu.test(line)) {84 if (/\bRegistered tunnel connection\b/iu.test(line)) {
73 return "connected";85 return "connected";
...@@ -105,6 +117,7 @@ export const make = Effect.gen(function* () {...@@ -105,6 +117,7 @@ export const make = Effect.gen(function* () {
105 const activeRef = yield* Ref.make<ActiveConnector | null>(null);117 const activeRef = yield* Ref.make<ActiveConnector | null>(null);
106 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);118 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
107 const reconcileSemaphore = yield* Semaphore.make(1);119 const reconcileSemaphore = yield* Semaphore.make(1);
120 const restartDelayRef = yield* Ref.make(0);
108 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];121 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
109122
110 const stopActive = Effect.gen(function* () {123 const stopActive = Effect.gen(function* () {
...@@ -115,6 +128,39 @@ export const make = Effect.gen(function* () {...@@ -115,6 +128,39 @@ export const make = Effect.gen(function* () {
115 const superviseConnector = (connector: ActiveConnector) =>128 const superviseConnector = (connector: ActiveConnector) =>
116 Effect.gen(function* () {129 Effect.gen(function* () {
117 const result = yield* Effect.result(connector.child.exitCode);130 const result = yield* Effect.result(connector.child.exitCode);
131 const activeAtExit = yield* Ref.get(activeRef);
132 if (
133 activeAtExit?.child.pid !== connector.child.pid ||
134 activeAtExit.configKey !== connector.configKey
135 ) {
136 return;
137 }
138 const uptimeMillis = (yield* Clock.currentTimeMillis) - connector.startedAtMillis;
139 // The first crash restarts immediately; every further crash inside the
140 // stable-uptime window doubles the wait, up to the cap. The delay runs
141 // before the semaphore so a user config change is never blocked behind
142 // it, and reconcileConfig re-checks the desired config afterwards.
143 const restartDelayMillis = yield* Ref.modify(restartDelayRef, (current) => {
144 if (uptimeMillis >= RELAY_RESTART_STABLE_UPTIME_MS) {
145 return [0, 0];
146 }
147 return [
148 current,
149 current === 0
150 ? RELAY_RESTART_BACKOFF_BASE_MS
151 : Math.min(current * 2, RELAY_RESTART_BACKOFF_MAX_MS),
152 ];
153 });
154 if (restartDelayMillis > 0) {
155 yield* Effect.logWarning("Relay client is crash-looping; delaying restart", {
156 pid: Number(connector.child.pid),
157 uptimeMillis,
158 restartDelayMillis,
159 tunnelId: connector.config.tunnelId,
160 tunnelName: connector.config.tunnelName,
161 });
162 yield* Effect.sleep(Duration.millis(restartDelayMillis));
163 }
118 yield* reconcileSemaphore.withPermits(1)(164 yield* reconcileSemaphore.withPermits(1)(
119 Effect.gen(function* () {165 Effect.gen(function* () {
120 const active = yield* Ref.get(activeRef);166 const active = yield* Ref.get(activeRef);
...@@ -274,6 +320,7 @@ export const make = Effect.gen(function* () {...@@ -274,6 +320,7 @@ export const make = Effect.gen(function* () {
274 scope: connectorScope,320 scope: connectorScope,
275 configKey: nextConfigKey,321 configKey: nextConfigKey,
276 config,322 config,
323 startedAtMillis: yield* Clock.currentTimeMillis,
277 } satisfies ActiveConnector;324 } satisfies ActiveConnector;
278 yield* Ref.set(activeRef, connector);325 yield* Ref.set(activeRef, connector);
279 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);326 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
...@@ -299,7 +346,11 @@ export const make = Effect.gen(function* () {...@@ -299,7 +346,11 @@ export const make = Effect.gen(function* () {
299 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(346 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(
300 (config: RelayManagedEndpointRuntimeConfig | null) =>347 (config: RelayManagedEndpointRuntimeConfig | null) =>
301 reconcileSemaphore.withPermits(1)(348 reconcileSemaphore.withPermits(1)(
302 Ref.set(desiredConfigRef, config).pipe(Effect.andThen(reconcileConfig(config))),349 // An explicit config change starts over with a fresh backoff.
350 Ref.set(restartDelayRef, 0).pipe(
351 Effect.andThen(Ref.set(desiredConfigRef, config)),
352 Effect.andThen(reconcileConfig(config)),
353 ),
303 ),354 ),
304 );355 );
305356