Skip to content

fix(server): wait for Cloudflare tunnel registration #8352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mateo-m wants to merge 6 commits into
base: main
Choose a base branch
from
+419 −102
Open

fix(server): wait for Cloudflare tunnel registration #8352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8b11a27
fix(server): wait for managed tunnel registration
mateo-m Aug 27, 2026
a2c3314
fix(server): keep relay diagnostics bounded
mateo-m Aug 27, 2026
b2cb866
fix(server): stop interrupted tunnel startup
mateo-m Aug 27, 2026
78b9eb4
fix(server): preserve relay connector recovery
mateo-m Aug 27, 2026
808dc81
fix(server): release relay restart permit
mateo-m Aug 27, 2026
14d6949
Merge branch 'main' into fix/cloudflared-readiness
mateo-m Aug 27, 2026
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
264 changes: 263 additions & 1 deletion apps/server/src/cloud/ManagedEndpointRuntime.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import * as Option from "effect/Option";
import * as PlatformError from "effect/PlatformError";
import * as Sink from "effect/Sink";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
import * as RelayClient from "@t3tools/shared/relayClient";

Expand Down Expand Up @@ -60,6 +61,7 @@ function makeHandle(input: {
readonly onKill: () => void;
readonly isRunning?: () => boolean;
readonly exitCode?: Effect.Effect<ChildProcessSpawner.ExitCode>;
readonly all?: ChildProcessSpawner.ChildProcessHandle["all"];
}) {
return ChildProcessSpawner.makeHandle({
pid: ChildProcessSpawner.ProcessId(input.pid),
Expand All @@ -73,7 +75,13 @@ function makeHandle(input: {
stdin: Sink.drain,
stdout: Stream.empty,
stderr: Stream.empty,
all: Stream.empty,
all:
input.all ??
Stream.make(
new TextEncoder().encode(
"2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n",
),
),
getInputFd: () => Sink.drain,
getOutputFd: () => Stream.empty,
});
Expand Down Expand Up @@ -283,6 +291,68 @@ describe("CloudManagedEndpointRuntime", () => {
}),
);

it.effect("does not block config changes while a restarted connector registers", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const firstExit = yield* Deferred.make<ChildProcessSpawner.ExitCode>();
const secondSpawned = yield* Deferred.make<void>();
const secondRegistration = yield* Deferred.make<void>();
let spawnCount = 0;
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
spawnCount += 1;
const pid = 410 + spawnCount;
if (spawnCount === 2) {
yield* Deferred.succeed(secondSpawned, undefined);
}
const handle = makeHandle({
pid,
...(spawnCount === 1
? {}
: {
all: Stream.fromEffect(Deferred.await(secondRegistration)).pipe(
Stream.map(() =>
new TextEncoder().encode(
"2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n",
),
),
),
}),
exitCode:
spawnCount === 1
? Deferred.await(firstExit)
: (Effect.never as Effect.Effect<ChildProcessSpawner.ExitCode>),
onKill: () => {
killed.push(pid);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
return handle;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(spawner);

yield* runtime.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token",
tunnelId: "tunnel-1",
});
yield* Deferred.succeed(firstExit, ChildProcessSpawner.ExitCode(1));
yield* Deferred.await(secondSpawned);

const stopFiber = yield* runtime.applyConfig(null).pipe(Effect.forkChild);
yield* Effect.yieldNow;
yield* Effect.yieldNow;
const stopped = stopFiber.pollUnsafe();
const killedAfterStop = [...killed];
yield* Deferred.succeed(secondRegistration, undefined);
yield* Fiber.join(stopFiber);

expect(stopped).toBeDefined();
expect(killedAfterStop).toEqual([411, 412]);
}),
);

it.effect("serializes concurrent connector config changes", () =>
Effect.gen(function* () {
const spawned: Array<number> = [];
Expand Down Expand Up @@ -333,6 +403,198 @@ describe("CloudManagedEndpointRuntime", () => {
}),
);

it.effect("does not report a running connector before Cloudflare registers it", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const registerConnection = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
const handle = makeHandle({
pid: 600,
all: Stream.fromEffect(Deferred.await(registerConnection)).pipe(
Stream.map(() =>
new TextEncoder().encode(
"2026-08-27T10:00:00Z INF Registered tunnel connection connIndex=0\n",
),
),
Stream.concat(Stream.never),
),
onKill: () => {
killed.push(600);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
return handle;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(spawner);

const statusFiber = yield* runtime
.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
})
.pipe(Effect.forkChild);
yield* Effect.yieldNow;

expect(statusFiber.pollUnsafe()).toBeUndefined();

yield* TestClock.adjust("15 seconds");
const status = yield* Fiber.join(statusFiber);

expect(status).toMatchObject({
status: "failed",
providerKind: "cloudflare_tunnel",
reason:
"Relay client did not register a tunnel connection within 15 seconds. Check whether the network allows outbound TCP and UDP traffic on port 7844.",
tunnelId: "tunnel-1",
});
expect(killed).toEqual([]);

yield* Deferred.succeed(registerConnection, undefined);
const recovered = yield* runtime.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
});

expect(recovered).toMatchObject({
status: "running",
pid: 600,
tunnelId: "tunnel-1",
});
expect(killed).toEqual([]);
}),
);

it.effect("restarts a connector that exits before Cloudflare registers it", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
let spawnCount = 0;
const secondSpawned = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
spawnCount += 1;
const pid = 600 + spawnCount;
const handle = makeHandle({
pid,
...(spawnCount === 1
? {
all: Stream.never,
exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)),
}
: {}),
onKill: () => {
killed.push(pid);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
if (spawnCount === 2) {
yield* Deferred.succeed(secondSpawned, undefined);
}
return handle;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(spawner);

const status = yield* runtime.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
});

expect(status).toMatchObject({
status: "failed",
providerKind: "cloudflare_tunnel",
reason: "Relay client exited before it registered a tunnel connection.",
tunnelId: "tunnel-1",
});
yield* Deferred.await(secondSpawned);
const recovered = yield* runtime.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
});

expect(recovered).toMatchObject({
status: "running",
providerKind: "cloudflare_tunnel",
pid: 602,
tunnelId: "tunnel-1",
});
expect(killed).toEqual([601]);
}),
);

it.effect("stops a connector when its first configuration is interrupted during spawn", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const processStarted = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
const handle = makeHandle({
pid: 602,
onKill: () => {
killed.push(602);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
yield* Deferred.succeed(processStarted, undefined);
return yield* Effect.never;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(spawner);

const statusFiber = yield* runtime
.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
})
.pipe(Effect.forkChild);
yield* Deferred.await(processStarted);
yield* Fiber.interrupt(statusFiber);

expect(killed).toEqual([602]);
}),
);

it.effect("stops a connector when its first configuration is interrupted", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const outputStarted = yield* Deferred.make<void>();
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
const handle = makeHandle({
pid: 602,
all: Stream.fromEffect(Deferred.succeed(outputStarted, undefined)).pipe(
Stream.flatMap(() => Stream.never),
),
onKill: () => {
killed.push(602);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
return handle;
}),
);
const runtime = yield* buildCloudManagedEndpointRuntime(spawner);

const statusFiber = yield* runtime
.applyConfig({
providerKind: "cloudflare_tunnel",
connectorToken: "token-secret",
tunnelId: "tunnel-1",
})
.pipe(Effect.forkChild);
yield* Deferred.await(outputStarted);
yield* Fiber.interrupt(statusFiber);

expect(killed).toEqual([602]);
}),
);

it.effect("reports connector spawn failures", () =>
Effect.gen(function* () {
const spawner = ChildProcessSpawner.make(() =>
Expand Down
Loading
Loading