Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 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
127 changes: 126 additions & 1 deletion apps/server/src/cloud/ManagedEndpointRuntime.test.ts
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 @@ -333,6 +341,123 @@ describe("CloudManagedEndpointRuntime", () => {
}),
);

it.effect("does not report a running connector before Cloudflare registers it", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
const handle = makeHandle({
pid: 600,
all: Stream.make(
new TextEncoder().encode(
"2026-08-27T10:00:00Z WRN Failed to dial edge with token-secret\n",
),
).pipe(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([600]);
}),
);

it.effect("reports a connector that exits before Cloudflare registers it", () =>
Effect.gen(function* () {
const killed: Array<number> = [];
const spawner = ChildProcessSpawner.make(() =>
Effect.gen(function* () {
const handle = makeHandle({
pid: 601,
all: Stream.never,
exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)),
onKill: () => {
killed.push(601);
},
});
yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore));
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",
});
expect(killed).toEqual([601]);
}),
);

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
59 changes: 44 additions & 15 deletions apps/server/src/cloud/ManagedEndpointRuntime.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay";
import * as RelayClient from "@t3tools/shared/relayClient";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Layer from "effect/Layer";
Expand All @@ -16,6 +17,8 @@ import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawne
import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
import { CLOUD_ENDPOINT_RUNTIME_CONFIG, decodeRuntimeConfig } from "./config.ts";

const RELAY_CONNECTION_TIMEOUT = "15 seconds";

function bytesToString(bytes: Uint8Array): string {
return new TextDecoder().decode(bytes);
}
Expand Down Expand Up @@ -63,6 +66,7 @@ export class CloudManagedEndpointRuntime extends Context.Service<

interface ActiveConnector {
readonly child: ChildProcessSpawner.ChildProcessHandle;
readonly connected: Deferred.Deferred<void>;
readonly scope: Scope.Closeable;
readonly configKey: string;
readonly config: RelayManagedEndpointRuntimeConfig;
Expand Down Expand Up @@ -112,6 +116,37 @@ export const make = Effect.gen(function* () {
yield* stopConnector(active);
});

const awaitConnectorConnection = Effect.fn(
"CloudManagedEndpointRuntime.awaitConnectorConnection",
)(function* (connector: ActiveConnector) {
const outcome = yield* Deferred.await(connector.connected).pipe(
Effect.as("connected" as const),
Effect.race(Effect.result(connector.child.exitCode).pipe(Effect.as("exited" as const))),
Effect.timeoutOption(RELAY_CONNECTION_TIMEOUT),
);
if (Option.isSome(outcome) && outcome.value === "connected") {
return {
status: "running",
providerKind: "cloudflare_tunnel",
pid: Number(connector.child.pid),
...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}),
...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}),
} satisfies CloudManagedEndpointRuntimeStatus;
}

yield* stopActive;
const reason = Option.isSome(outcome)
? "Relay client exited before it registered a tunnel connection."
: `Relay client did not register a tunnel connection within ${RELAY_CONNECTION_TIMEOUT}. Check whether the network allows outbound TCP and UDP traffic on port 7844.`;
return {
status: "failed",
providerKind: "cloudflare_tunnel",
reason,
...(connector.config.tunnelId ? { tunnelId: connector.config.tunnelId } : {}),
...(connector.config.tunnelName ? { tunnelName: connector.config.tunnelName } : {}),
} satisfies CloudManagedEndpointRuntimeStatus;
Comment thread
mateo-m marked this conversation as resolved.
Outdated
});

const superviseConnector = (connector: ActiveConnector) =>
Effect.gen(function* () {
const result = yield* Effect.result(connector.child.exitCode);
Expand Down Expand Up @@ -167,7 +202,11 @@ export const make = Effect.gen(function* () {
};
switch (classifyRelayClientOutput(line)) {
case "connected":
return Effect.logInfo("Relay client tunnel connection registered", attributes);
return Deferred.succeed(connector.connected, undefined).pipe(
Effect.andThen(
Effect.logInfo("Relay client tunnel connection registered", attributes),
),
);
case "warning":
return Effect.logWarning("Relay client reported a transport warning", attributes);
case "debug":
Expand Down Expand Up @@ -197,13 +236,7 @@ export const make = Effect.gen(function* () {
if (active?.configKey === nextConfigKey) {
const isRunning = yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false));
if (isRunning) {
return {
status: "running",
providerKind: "cloudflare_tunnel",
pid: Number(active.child.pid),
...(active.config.tunnelId ? { tunnelId: active.config.tunnelId } : {}),
...(active.config.tunnelName ? { tunnelName: active.config.tunnelName } : {}),
} satisfies CloudManagedEndpointRuntimeStatus;
return yield* awaitConnectorConnection(active);
}
}

Expand Down Expand Up @@ -269,22 +302,18 @@ export const make = Effect.gen(function* () {
}

if (!("status" in child)) {
const connected = yield* Deferred.make<void>();
const connector = {
child,
connected,
scope: connectorScope,
configKey: nextConfigKey,
config,
} satisfies ActiveConnector;
yield* Ref.set(activeRef, connector);
yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
yield* Effect.forkIn(superviseConnector(connector), connectorScope);
return {
status: "running",
providerKind: "cloudflare_tunnel",
pid: Number(child.pid),
...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
} satisfies CloudManagedEndpointRuntimeStatus;
return yield* awaitConnectorConnection(connector).pipe(Effect.onInterrupt(() => stopActive));
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}

return {
Expand Down
5 changes: 4 additions & 1 deletion apps/server/src/cloud/http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -467,7 +467,10 @@ const applyCloudRelayConfig = Effect.fn("environment.cloud.applyRelayConfig")(fu
endpointRuntimeStatus.status === "disabled" || endpointRuntimeStatus.status === "running";
if (!ok) {
return yield* new EnvironmentCloudEndpointUnavailableError({
message: "Managed endpoint runtime could not be started.",
message:
endpointRuntimeStatus.status === "failed"
? `Managed endpoint runtime could not connect. ${endpointRuntimeStatus.reason}`
: "Managed endpoint runtime could not connect.",
endpointRuntimeStatus,
});
}
Expand Down
5 changes: 4 additions & 1 deletion apps/server/src/server.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2962,7 +2962,10 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
endpointRuntimeStatus?: { status?: string; reason?: string };
}>(relayConfigResponse);
assert.equal(relayConfigBody._tag, "EnvironmentCloudEndpointUnavailableError");
assert.equal(relayConfigBody.message, "Managed endpoint runtime could not be started.");
assert.equal(
relayConfigBody.message,
"Managed endpoint runtime could not connect. cloudflared missing",
);
assert.equal(relayConfigBody.endpointRuntimeStatus?.status, "failed");
assert.equal(relayConfigBody.endpointRuntimeStatus?.reason, "cloudflared missing");

Expand Down
Loading