import { AuthStandardClientScopes, EnvironmentId } from "@t3tools/contracts";import { RelayClientTracer } from "@t3tools/shared/relayTracing";import { describe, expect, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import * as TestClock from "effect/testing/TestClock";import * as Tracer from "effect/Tracer"; import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";import * as TokenStore from "../authorization/tokenStore.ts";import * as ClientCapabilities from "../platform/capabilities.ts";import {  ManagedRelayClient,  ManagedRelayDpopSigner,  ManagedRelayRequestTimeoutError,} from "../relay/managedRelay.ts";import { remoteHttpClientLayer } from "../rpc/http.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import { fetchEnvironmentSessionState } from "../state/session.ts";import type { ConnectionCatalogEntry } from "./catalog.ts";import * as Connectivity from "./connectivity.ts";import * as ConnectionDriver from "./driver.ts";import {  ConnectionBlockedError,  ConnectionTransientError,  PrimaryConnectionTarget,  RelayConnectionTarget,  type ConnectionAttemptError,  type ConnectionTarget,  type NetworkStatus,  type PreparedConnection,  type SupervisorConnectionState,} from "./model.ts";import * as RpcSession from "../rpc/session.ts";import * as EnvironmentSupervisor from "./supervisor.ts";import * as ConnectionWakeups from "./wakeups.ts";import { NETWORK_BLOCKING_HINT } from "../errors/network.ts"; const TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-1"),  label: "Test environment",  httpBaseUrl: "https://environment.example.test",  wsBaseUrl: "wss://environment.example.test",}); const RELAY_TARGET = new RelayConnectionTarget({  environmentId: TARGET.environmentId,  label: TARGET.label,}); const TARGET_ENTRY: ConnectionCatalogEntry = {  target: TARGET,  profile: Option.none(),}; const RELAY_ENTRY: ConnectionCatalogEntry = {  target: RELAY_TARGET,  profile: Option.none(),}; const PREPARED_CONNECTION: PreparedConnection = {  environmentId: TARGET.environmentId,  label: TARGET.label,  httpBaseUrl: TARGET.httpBaseUrl,  socketUrl: "wss://environment.example.test/ws",  httpAuthorization: null,  target: TARGET,}; const TEST_RPC_CLIENT = {} as WsRpcProtocolClient; function transient(message = "Connection failed.") {  return new ConnectionTransientError({    reason: "transport",    detail: message,  });} function blocked(message = "Authentication required.") {  return new ConnectionBlockedError({    reason: "authentication",    detail: message,  });} function awaitState(  state: SubscriptionRef.SubscriptionRef<SupervisorConnectionState>,  predicate: (value: SupervisorConnectionState) => boolean,) {  return SubscriptionRef.changes(state).pipe(    Stream.filter(predicate),    Stream.runHead,    Effect.map(Option.getOrThrow),  );} const eventuallyState = Effect.fn("TestConnectionHarness.eventuallyState")(function* (  state: SubscriptionRef.SubscriptionRef<SupervisorConnectionState>,  predicate: (value: SupervisorConnectionState) => boolean,) {  let lastState = yield* SubscriptionRef.get(state);  for (let iteration = 0; iteration < 100; iteration += 1) {    lastState = yield* SubscriptionRef.get(state);    if (predicate(lastState)) {      return lastState;    }    yield* Effect.yieldNow;  }  return yield* Effect.die(    new Error(      `Expected supervisor state was not observed. Last state: phase=${lastState.phase}, stage=${lastState.stage ?? "none"}, attempt=${lastState.attempt}, generation=${lastState.generation}`,    ),  );}); const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: {  readonly networkStatus?: NetworkStatus;  readonly prepare?: (    attempt: number,    target: ConnectionTarget,  ) => Effect.Effect<PreparedConnection, ConnectionAttemptError>;  readonly ready?: (attempt: number) => Effect.Effect<void, ConnectionAttemptError>;  readonly probe?: (attempt: number) => Effect.Effect<void, ConnectionAttemptError>;}) {  const networkStatus = yield* SubscriptionRef.make<NetworkStatus>(    options?.networkStatus ?? "online",  );  const prepareCount = yield* Ref.make(0);  const sessionCount = yield* Ref.make(0);  const releaseCount = yield* Ref.make(0);  const wakeups = yield* SubscriptionRef.make<{    readonly sequence: number;    readonly reason: ConnectionWakeups.ConnectionWakeup;  }>({    sequence: 0,    reason: "application-active",  });  const closedSessions = yield* Ref.make<    ReadonlyArray<Deferred.Deferred<never, ConnectionTransientError>>  >([]);   const connectivity = Connectivity.Connectivity.of({    status: SubscriptionRef.get(networkStatus),    changes: SubscriptionRef.changes(networkStatus),  });   const prepare = Effect.fn("TestConnectionDriver.prepare")(function* (target: ConnectionTarget) {    const attempt = yield* Ref.updateAndGet(prepareCount, (count) => count + 1);    if (options?.prepare) {      return yield* options.prepare(attempt, target);    }    return PREPARED_CONNECTION;  });   const connect = Effect.fn("TestConnectionDriver.connect")(function* (    entry: ConnectionCatalogEntry,    reportProgress: (progress: ConnectionDriver.ConnectionDriverProgress) => Effect.Effect<void>,  ) {    const target = entry.target;    yield* reportProgress({ stage: "preparing" });    const prepared = yield* prepare(target);    yield* reportProgress({ stage: "opening", prepared });     const attempt = yield* Ref.updateAndGet(sessionCount, (count) => count + 1);    const closed = yield* Deferred.make<never, ConnectionTransientError>();    yield* Ref.update(closedSessions, (sessions) => [...sessions, closed]);     const session = yield* Effect.acquireRelease(      Effect.succeed({        client: TEST_RPC_CLIENT,        initialConfig: Effect.die(new Error("Initial config is not used by supervisor tests.")),        subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input),        ready: options?.ready?.(attempt) ?? Effect.void,        probe: options?.probe?.(attempt) ?? Effect.void,        closed: Deferred.await(closed),      } satisfies RpcSession.RpcSession),      () => Ref.update(releaseCount, (count) => count + 1),    );     yield* reportProgress({ stage: "synchronizing", prepared });    yield* session.ready;    return { prepared, session } satisfies ConnectionDriver.EnvironmentConnectionLease;  });   const dependencies = Layer.mergeAll(    Layer.succeed(Connectivity.Connectivity, connectivity),    Layer.succeed(      ConnectionWakeups.ConnectionWakeups,      ConnectionWakeups.ConnectionWakeups.of({        changes: SubscriptionRef.changes(wakeups).pipe(          Stream.drop(1),          Stream.map((event) => event.reason),        ),      }),    ),    Layer.succeed(      ConnectionDriver.ConnectionDriver,      ConnectionDriver.ConnectionDriver.of({ connect }),    ),  );   return {    dependencies,    prepareCount,    sessionCount,    releaseCount,    setNetworkStatus: (status: NetworkStatus) => SubscriptionRef.set(networkStatus, status),    wake: (reason: ConnectionWakeups.ConnectionWakeup) =>      SubscriptionRef.update(wakeups, (event) => ({        sequence: event.sequence + 1,        reason,      })),    closeLatestSession: Effect.fn("TestConnectionHarness.closeLatestSession")(function* (      error = transient("Session closed."),    ) {      const sessions = yield* Ref.get(closedSessions);      const latest = sessions.at(-1);      if (latest) {        yield* Deferred.fail(latest, error);      }    }),  };}); describe("EnvironmentSupervisor", () => {  it.effect("exports each relay setup as a standalone linked trace that ends at readiness", () =>    Effect.gen(function* () {      const spans: Array<Tracer.NativeSpan> = [];      const tracer = Tracer.make({        span: (options) => {          const span = new Tracer.NativeSpan(options);          const end = span.end.bind(span);          span.end = (endTime, exit) => {            end(endTime, exit);            spans.push(span);          };          return span;        },      });      const harness = yield* makeHarness({        prepare: (attempt) =>          attempt === 1 ? Effect.fail(transient()) : Effect.succeed(PREPARED_CONNECTION),      });      const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, {        initiallyDesired: true,      }).pipe(        Effect.provide(harness.dependencies),        Effect.provideService(RelayClientTracer, Option.some(tracer)),      );       yield* awaitState(        supervisor.state,        (state) => state.phase === "backoff" && state.attempt === 1,      );      const firstAttempt = spans.find((span) => span.name === "relay.connection.attempt");      expect(firstAttempt).toBeDefined();       yield* TestClock.adjust("3 seconds");      yield* awaitState(supervisor.state, (state) => state.phase === "connected");       const attempts = spans.filter((span) => span.name === "relay.connection.attempt");      expect(attempts).toHaveLength(2);      expect(attempts[0]?.traceId).not.toBe(attempts[1]?.traceId);      expect(attempts[1]?.links.map((link) => link.span.spanId)).toContain(attempts[0]?.spanId);      expect(yield* Ref.get(harness.releaseCount)).toBe(0);    }).pipe(Effect.provide(TestClock.layer())),  );   it.effect("does not attempt a connection until it is desired", () =>    Effect.gen(function* () {      const harness = yield* makeHarness();      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY).pipe(        Effect.provide(harness.dependencies),      );       expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("available");      expect(yield* Ref.get(harness.prepareCount)).toBe(0);    }),  );   it.effect("does not let the initial connect signal cancel the first attempt", () =>    Effect.gen(function* () {      const harness = yield* makeHarness();      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY).pipe(        Effect.provide(harness.dependencies),      );       yield* supervisor.connect;      yield* awaitState(supervisor.state, (state) => state.phase === "connected");       expect(yield* Ref.get(harness.sessionCount)).toBe(1);      expect(yield* Ref.get(harness.releaseCount)).toBe(0);    }),  );   it.effect("waits while offline and connects immediately when the network returns", () =>    Effect.gen(function* () {      const harness = yield* makeHarness({ networkStatus: "offline" });      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(supervisor.state, (state) => state.phase === "offline");      expect(yield* Ref.get(harness.prepareCount)).toBe(0);       yield* harness.setNetworkStatus("online");      const ready = yield* awaitState(supervisor.state, (state) => state.phase === "connected");       expect(ready).toMatchObject({        desired: true,        network: "online",        phase: "connected",        attempt: 1,        generation: 1,        lastFailure: null,      });      expect(yield* Ref.get(harness.prepareCount)).toBe(1);    }),  );   it.effect("resets retries when activation arrives before the network returns", () =>    Effect.gen(function* () {      const harness = yield* makeHarness();      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(supervisor.state, (state) => state.phase === "connected");      yield* harness.closeLatestSession();      yield* awaitState(        supervisor.state,        (state) => state.phase === "backoff" && state.attempt === 1,      );      yield* harness.setNetworkStatus("offline");      yield* awaitState(        supervisor.state,        (state) => state.phase === "offline" && state.attempt === 2,      );       yield* harness.wake("application-active-reconnect");      yield* awaitState(        supervisor.state,        (state) => state.phase === "offline" && state.attempt === 1,      );      yield* harness.setNetworkStatus("online");      yield* awaitState(        supervisor.state,        (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1,      );    }),  );   it.effect("retries forever with exponential backoff capped at sixteen seconds", () =>    Effect.gen(function* () {      const harness = yield* makeHarness({        prepare: () => Effect.fail(transient()),      });      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(        supervisor.state,        (state) => state.phase === "backoff" && state.attempt === 1,      );      expect(yield* Ref.get(harness.prepareCount)).toBe(1);       for (const [index, delay] of [3_000, 4_000, 8_000, 16_000, 16_000, 16_000].entries()) {        yield* TestClock.adjust(delay);        yield* eventuallyState(          supervisor.state,          (state) => state.phase === "backoff" && state.attempt === index + 2,        );      }       expect(yield* Ref.get(harness.prepareCount)).toBe(7);    }).pipe(Effect.provide(TestClock.layer())),  );   it.effect("keeps the latest failure visible throughout the next connection attempt", () =>    Effect.gen(function* () {      const harness = yield* makeHarness({        prepare: (attempt) =>          attempt === 1 ? Effect.fail(transient("Relay connection timed out.")) : Effect.never,      });      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(        supervisor.state,        (state) => state.phase === "backoff" && state.attempt === 1,      );      yield* TestClock.adjust("3 seconds");       const retrying = yield* awaitState(        supervisor.state,        (state) =>          state.phase === "connecting" && state.stage === "preparing" && state.attempt === 2,      );      expect(retrying).toMatchObject({        phase: "connecting",        stage: "preparing",        attempt: 2,        lastFailure: {          _tag: "ConnectionTransientError",          reason: "transport",          message: "Relay connection timed out.",        },      });    }).pipe(Effect.provide(TestClock.layer())),  );   it.effect("retries when a session never becomes ready", () =>    Effect.gen(function* () {      const harness = yield* makeHarness({        ready: () => Effect.never,      });      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(        supervisor.state,        (state) => state.phase === "connecting" && state.stage === "synchronizing",      );      yield* TestClock.adjust("14 seconds");      expect((yield* SubscriptionRef.get(supervisor.state)).stage).toBe("synchronizing");       yield* TestClock.adjust("1 second");      const retrying = yield* awaitState(supervisor.state, (state) => state.phase === "backoff");       expect(retrying).toMatchObject({        phase: "backoff",        lastFailure: {          _tag: "ConnectionTransientError",          reason: "timeout",          message: "Test environment did not respond during connection setup.",        },      });      expect(yield* Ref.get(harness.releaseCount)).toBe(1);      expect(Option.isNone(yield* SubscriptionRef.get(supervisor.prepared))).toBe(true);    }).pipe(Effect.provide(TestClock.layer())),  );   it.effect("interrupts and releases a connection attempt when setup times out", () =>    Effect.gen(function* () {      const harness = yield* makeHarness({        prepare: () => Effect.never,      });      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {        initiallyDesired: true,      }).pipe(Effect.provide(harness.dependencies));       yield* awaitState(        supervisor.state,        (state) => state.phase === "connecting" && state.stage === "preparing",      );      yield* TestClock.adjust("15 seconds");      const retrying = yield* eventuallyState(        supervisor.state,        (state) => state.phase === "backoff" && state.attempt === 1,      );       expect(retrying).toMatchObject({        lastFailure: {          _tag: "ConnectionTransientError",          reason: "timeout",          message: "Test environment did not respond during connection setup.",        },      });    }).pipe(Effect.provide(TestClock.layer())),  );   it.effect(    "shows a network hint for a stalled relay connection and clears it after recovery",    () =>      Effect.gen(function* () {        const harness = yield* makeHarness({          prepare: (attempt) =>            attempt === 1 ? Effect.never : Effect.succeed(PREPARED_CONNECTION),        });        const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, {          initiallyDesired: true,        }).pipe(Effect.provide(harness.dependencies));         yield* awaitState(supervisor.state, (state) => state.phase === "connecting");        yield* TestClock.adjust("15 seconds");        const failed = yield* awaitState(supervisor.state, (state) => state.phase === "backoff");        expect(failed.lastFailure?.message).toBe(          `Test environment did not respond during connection setup. ${NETWORK_BLOCKING_HINT}`,        );         yield* TestClock.adjust("3 seconds");        const recovered = yield* awaitState(

Lines 1–500 of 1418. Load more