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
127 changes: 123 additions & 4 deletions apps/server/src/cloud/ManagedEndpointRuntime.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import * as NodeServices from "@effect/platform-node/NodeServices";
import { describe, expect, it } from "@effect/vitest";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
Expand All @@ -19,6 +21,7 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay
import * as RelayClient from "@t3tools/shared/relayClient";

import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
import * as ServerConfig from "../config.ts";
import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";

const layerRelayClientAvailable = Layer.succeed(
Expand All @@ -38,29 +41,34 @@ const layerRelayClientAvailable = Layer.succeed(

const unusedHttpClient = HttpClient.make(() => Effect.die("unused"));

const emptySecretStoreLayer = Layer.mock(ServerSecretStore.ServerSecretStore)({
get: () => Effect.succeedNone,
set: () => Effect.void,
});

const runtimeDependencies = (
spawner: ReturnType<typeof ChildProcessSpawner.make>,
relayClientLayer = layerRelayClientAvailable,
httpClient = unusedHttpClient,
secretStoreLayer = emptySecretStoreLayer,
) =>
Layer.mergeAll(
Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner),
Layer.succeed(HttpClient.HttpClient, httpClient),
relayClientLayer,
Layer.mock(ServerSecretStore.ServerSecretStore)({
get: () => Effect.succeedNone,
}),
secretStoreLayer,
);

const buildCloudManagedEndpointRuntime = (
spawner: ReturnType<typeof ChildProcessSpawner.make>,
relayClientLayer = layerRelayClientAvailable,
httpClient = unusedHttpClient,
secretStoreLayer = emptySecretStoreLayer,
) =>
Effect.gen(function* () {
const context = yield* Layer.build(
ManagedEndpointRuntime.layer.pipe(
Layer.provide(runtimeDependencies(spawner, relayClientLayer, httpClient)),
Layer.provide(runtimeDependencies(spawner, relayClientLayer, httpClient, secretStoreLayer)),
),
);
return yield* Effect.service(ManagedEndpointRuntime.CloudManagedEndpointRuntime).pipe(
Expand Down Expand Up @@ -403,6 +411,117 @@ describe("CloudManagedEndpointRuntime", () => {
}),
);

it.effect(
"retains automatic HTTP/2 fallback across restarts until expiry and respects overrides",
() =>
Effect.gen(function* () {
const context = yield* Layer.build(
ServerSecretStore.layer.pipe(
Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-transport-test-" })),
Layer.provide(NodeServices.layer),
),
);
const secretStoreLayer = Layer.succeed(
ServerSecretStore.ServerSecretStore,
context.pipe(Context.get(ServerSecretStore.ServerSecretStore)),
);
const fixture = yield* makeTransportSpawner();
const start = (spawner = fixture.spawner) =>
buildCloudManagedEndpointRuntime(
spawner,
layerRelayClientAvailable,
unusedHttpClient,
secretStoreLayer,
);
yield* Effect.gen(function* () {
const runtime = yield* start();
yield* runtime.applyConfig(tunnelConfig);
expect(yield* Queue.take(fixture.started)).toBe(0);
expect(fixture.spawned[0]!.args).not.toContain("--protocol");
yield* fixture.emit(0, quicDialWarning.repeat(30));
expect(yield* Queue.take(fixture.started)).toBe(1);
yield* fixture.observe(runtime, 1);
expect(fixture.spawned[1]!.args.slice(-3)).toEqual(["--protocol", "http2", "run"]);
}).pipe(Effect.scoped);

yield* TestClock.adjust(Duration.hours(1));
for (const transport of [undefined, "quic", "http2", "auto", undefined]) {
yield* Effect.gen(function* () {
const next = yield* makeTransportSpawner(transport ? { transport } : {});
const restarted = yield* start(next.spawner);
yield* restarted.applyConfig(tunnelConfig);
if (transport === undefined) {
expect(next.spawned[0]!.args.slice(-3)).toEqual(["--protocol", "http2", "run"]);
} else {
expect(next.spawned[0]!.args).not.toContain("--protocol");
expect(next.spawned[0]!.options.env?.TUNNEL_TRANSPORT_PROTOCOL).toBe(transport);
yield* next.observe(restarted, 0, quicDialWarning.repeat(30));
expect(next.spawned).toHaveLength(1);
}
}).pipe(Effect.scoped);
}

// Reusing the hint one hour later must not extend its original lifetime.
yield* TestClock.adjust(Duration.hours(23));
yield* Effect.gen(function* () {
const next = yield* makeTransportSpawner();
const restarted = yield* start(next.spawner);
yield* restarted.applyConfig(tunnelConfig);
expect(next.spawned[0]!.args).not.toContain("--protocol");
expect(yield* Queue.take(next.started)).toBe(0);
yield* next.emit(0, quicDatagramWarning.repeat(30));
expect(yield* Queue.take(next.started)).toBe(1);
expect(next.spawned[1]!.args.slice(-3)).toEqual(["--protocol", "http2", "run"]);
}).pipe(Effect.scoped);
}),
);

it.effect(
"ignores invalid or unreadable fallback state and still falls back if persistence fails",
() =>
Effect.gen(function* () {
for (const cached of [
"not json",
'{"fellBackAt":"invalid"}',
'{"fellBackAt":9007199254740991}',
null,
]) {
yield* Effect.gen(function* () {
const fixture = yield* makeTransportSpawner();
const runtime = yield* buildCloudManagedEndpointRuntime(
fixture.spawner,
layerRelayClientAvailable,
unusedHttpClient,
Layer.mock(ServerSecretStore.ServerSecretStore)({
get: () =>
cached === null
? Effect.fail(
new ServerSecretStore.SecretStoreReadError({
resource: "test",
cause: "unreadable",
}),
)
: Effect.succeedSome(new TextEncoder().encode(cached)),
set: () =>
Effect.fail(
new ServerSecretStore.SecretStorePersistError({
resource: "test",
cause: "read-only",
}),
),
}),
);
yield* runtime.applyConfig(tunnelConfig);
expect(yield* Queue.take(fixture.started)).toBe(0);
expect(fixture.spawned[0]!.args).not.toContain("--protocol");
yield* fixture.emit(0, quicDialWarning.repeat(30));
expect(yield* Queue.take(fixture.started)).toBe(1);
expect(fixture.spawned[1]!.args.slice(-3)).toEqual(["--protocol", "http2", "run"]);
}).pipe(Effect.scoped);
}
}),
);

it.effect("requests recovery after a failed HTTP/2 spawn and retries with HTTP/2", () =>
Effect.gen(function* () {
const fixture = yield* makeTransportSpawner({ failSpawn: 1 });
Expand Down
32 changes: 31 additions & 1 deletion apps/server/src/cloud/ManagedEndpointRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ import { HttpClient, HttpClientResponse } from "effect/http";
import * as ChildProcess from "effect/process/ChildProcess";
import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";

import * as ServerSecretStore from "../auth/ServerSecretStore.ts";

export type CloudManagedEndpointRuntimeStatus =
| {
readonly status: "disabled";
Expand Down Expand Up @@ -109,6 +111,12 @@ const ConnectorConfigBody = Schema.Struct({
// immediately. Keep that evidence across registrations and connector restarts.
const QUIC_STALLS_BEFORE_HTTP2 = 30;
const QUIC_STALL_WINDOW_MS = 5 * 60_000;
// Retry QUIC on a later startup after transient failures or network changes; reuse never renews this hint.
const HTTP2_FALLBACK_TTL_MS = 24 * 60 * 60_000;
const HTTP2_FALLBACK_STATE = "cloud-endpoint-http2-fallback";
const Http2Fallback = Schema.fromJsonString(Schema.Struct({ fellBackAt: Schema.Number }));
const decodeHttp2Fallback = Schema.decodeUnknownOption(Http2Fallback);
const encodeHttp2Fallback = Schema.encodeSync(Http2Fallback);
// A failed background install of the pinned relay client is retried no sooner
// than this, so a crash-looping connector or an offline host does not redownload
// on every reconcile.
Expand Down Expand Up @@ -182,6 +190,7 @@ const stopConnector = (connector: ActiveConnector | null) =>
export const make = Effect.gen(function* () {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
const relayClient = yield* RelayClient.RelayClient;
const secrets = yield* ServerSecretStore.ServerSecretStore;
const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
const activeRef = yield* Ref.make<ActiveConnector | null>(null);
const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);
Expand All @@ -199,7 +208,18 @@ export const make = Effect.gen(function* () {
const notReadyChecksBeforeRestartRef = yield* Ref.make(CONNECTOR_NOT_READY_CHECKS_BEFORE_RESTART);
// An explicit cloudflared transport in the environment is the user's choice.
const transportChosenByUser = process.env.TUNNEL_TRANSPORT_PROTOCOL !== undefined;
const http2Ref = yield* Ref.make(false);
const previousFallback = transportChosenByUser
? Option.none()
: yield* secrets.get(HTTP2_FALLBACK_STATE).pipe(
Effect.map(Option.flatMap((bytes) => decodeHttp2Fallback(new TextDecoder().decode(bytes)))),
Effect.orElseSucceed(() => Option.none()),
);
const now = yield* Clock.currentTimeMillis;
const http2Ref = yield* Ref.make(
Option.isSome(previousFallback) &&
now >= previousFallback.value.fellBackAt &&
now - previousFallback.value.fellBackAt < HTTP2_FALLBACK_TTL_MS,
);
const quicStallsRef = yield* Ref.make<ReadonlyArray<number>>([]);
let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];

Expand Down Expand Up @@ -452,6 +472,16 @@ export const make = Effect.gen(function* () {
const desiredConfig = yield* Ref.get(desiredConfigRef);
if ((yield* Ref.get(http2Ref)) || active !== connector || !desiredConfig) return;
yield* Ref.set(http2Ref, true);
yield* secrets
.set(
HTTP2_FALLBACK_STATE,
new TextEncoder().encode(
encodeHttp2Fallback({ fellBackAt: yield* Clock.currentTimeMillis }),
),
)
.pipe(
Effect.catch(() => Effect.logWarning("Could not persist the relay client HTTP/2 fallback")),
);
yield* Effect.logWarning("Relay client keeps losing QUIC connections; switching to HTTP/2", {
pid: Number(connector.child.pid),
tunnelId: connector.config.tunnelId,
Expand Down
Loading