import { EnvironmentId, type RelayClientInstallProgressEvent, WS_METHODS,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Option from "effect/Option";import * as Queue from "effect/Queue";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 { RpcClientError } from "effect/unstable/rpc"; import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection, type SupervisorConnectionState,} from "../connection/model.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import * as RpcSession from "../rpc/session.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import { EnvironmentRpcRequestObserver, request, runStream, subscribe } from "./client.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment", httpBaseUrl: "https://environment.example.test", wsBaseUrl: "wss://environment.example.test",}); const INSTALL_CHECKING: RelayClientInstallProgressEvent = { type: "progress", stage: "checking",};const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = { type: "progress", stage: "downloading",}; function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, initialConfig: Effect.never, ready: Effect.void, probe: Effect.void, closed: Effect.never, };} const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () { const state = yield* SubscriptionRef.make<SupervisorConnectionState>(AVAILABLE_CONNECTION_STATE); const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>( Option.none(), ); const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()); const retryCount = yield* Ref.make(0); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state, session: activeSession, prepared, connect: Effect.void, disconnect: Effect.void, retryNow: Ref.update(retryCount, (count) => count + 1), } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); return { activeSession, retryCount, supervisor, };}); describe("environment RPC", () => { it.effect("observes unary requests until they complete", () => Effect.gen(function* () { const observations: string[] = []; const client = { [WS_METHODS.cloudGetRelayClientStatus]: () => Effect.succeed({ status: "available", version: "2026.6.0" }), } as unknown as WsRpcProtocolClient; const { activeSession, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(client))); const result = yield* request(WS_METHODS.cloudGetRelayClientStatus, {}).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService( EnvironmentRpcRequestObserver, EnvironmentRpcRequestObserver.of({ observe: ({ environmentId, method }) => Effect.sync(() => { observations.push(`start:${environmentId}:${method}`); return Effect.sync(() => { observations.push(`finish:${environmentId}:${method}`); }); }), }), ), ); expect(result).toEqual({ status: "available", version: "2026.6.0" }); expect(observations).toEqual([ `start:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`, `finish:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`, ]); }), ); it.effect("binds finite streaming commands to one active session", () => Effect.gen(function* () { const firstEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>(); const secondEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>(); const firstClient = { [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(firstEvents), } as unknown as WsRpcProtocolClient; const secondClient = { [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(secondEvents), } as unknown as WsRpcProtocolClient; const { activeSession, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient))); const resultFiber = yield* runStream(WS_METHODS.cloudInstallRelayClient, {}).pipe( Stream.take(2), Stream.runCollect, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.forkChild, ); yield* Effect.yieldNow; yield* Queue.offer(firstEvents, INSTALL_CHECKING); yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient))); yield* Queue.offer(secondEvents, INSTALL_DOWNLOADING); yield* Queue.offer(firstEvents, INSTALL_DOWNLOADING); expect(yield* Fiber.join(resultFiber)).toEqual([INSTALL_CHECKING, INSTALL_DOWNLOADING]); }), ); it.effect("switches durable subscriptions when the supervisor replaces the session", () => Effect.gen(function* () { const subscriptions: string[] = []; const firstClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("first"); return Stream.never; }, } as unknown as WsRpcProtocolClient; const secondClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("second"); return Stream.never; }, } as unknown as WsRpcProtocolClient; const { activeSession, retryCount, supervisor } = yield* makeHarness(); const awaitSubscriptions = Effect.fn("TestEnvironmentRpc.awaitSubscriptions")(function* ( count: number, ) { for (let attempt = 0; attempt < 100; attempt += 1) { if (subscriptions.length >= count) { return; } yield* Effect.yieldNow; } return yield* Effect.die(new Error(`Expected ${count} durable subscriptions.`)); }); const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.forkChild, ); yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient))); yield* awaitSubscriptions(1); yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient))); yield* awaitSubscriptions(2); yield* Fiber.interrupt(subscriptionFiber); expect(subscriptions).toEqual(["first", "second"]); expect(yield* Ref.get(retryCount)).toBe(0); }), ); it.effect("keeps durable subscriptions alive across a transport failure and new session", () => Effect.gen(function* () { const subscriptions: string[] = []; const firstClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("first"); return Stream.fail( new RpcClientError.RpcClientError({ reason: new RpcClientError.RpcClientDefect({ message: "socket closed", cause: new Error("socket closed"), }), }), ); }, } as unknown as WsRpcProtocolClient; const secondClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("second"); return Stream.never; }, } as unknown as WsRpcProtocolClient; const { activeSession, retryCount, supervisor } = yield* makeHarness(); const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.forkChild, ); yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient))); for (let attempt = 0; attempt < 100 && subscriptions.length < 1; attempt += 1) { yield* Effect.yieldNow; } yield* SubscriptionRef.set(activeSession, Option.none()); yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient))); for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) { yield* Effect.yieldNow; } yield* Fiber.interrupt(subscriptionFiber); expect(subscriptions).toEqual(["first", "second"]); expect(yield* Ref.get(retryCount)).toBe(0); }), ); it.effect("surfaces domain subscription failures without reconnecting", () => Effect.gen(function* () { const domainError = new Error("terminal subscription rejected"); const client = { [WS_METHODS.subscribeTerminalEvents]: () => Stream.fail(domainError), } as unknown as WsRpcProtocolClient; const { activeSession, retryCount, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(client))); const error = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.flip, ); expect(error).toBe(domainError); expect(yield* Ref.get(retryCount)).toBe(0); }), ); it.effect("keeps handled domain failures dormant until a replacement session arrives", () => Effect.gen(function* () { const domainError = new Error("terminal subscription rejected"); const subscriptions: string[] = []; const observedFailures: Error[] = []; const firstClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("first"); return Stream.fail(domainError); }, } as unknown as WsRpcProtocolClient; const secondClient = { [WS_METHODS.subscribeTerminalEvents]: () => { subscriptions.push("second"); return Stream.never; }, } as unknown as WsRpcProtocolClient; const { activeSession, retryCount, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient))); const subscriptionFiber = yield* subscribe( WS_METHODS.subscribeTerminalEvents, {}, { onExpectedFailure: (cause) => Effect.sync(() => { observedFailures.push(Cause.squash(cause) as Error); }), }, ).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.forkChild, ); for (let attempt = 0; attempt < 100 && observedFailures.length < 1; attempt += 1) { yield* Effect.yieldNow; } expect(subscriptions).toEqual(["first"]); expect(observedFailures).toEqual([domainError]); yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient))); for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) { yield* Effect.yieldNow; } yield* Fiber.interrupt(subscriptionFiber); expect(subscriptions).toEqual(["first", "second"]); expect(yield* Ref.get(retryCount)).toBe(0); }), ); it.effect("retries handled domain failures within the same session when configured", () => Effect.gen(function* () { const domainError = new Error("thread not found yet"); const subscriptionCount = yield* Ref.make(0); const expectedFailureCount = yield* Ref.make(0); const client = { [WS_METHODS.subscribeTerminalEvents]: () => Stream.unwrap( Ref.getAndUpdate(subscriptionCount, (count) => count + 1).pipe( Effect.map((count) => (count === 0 ? Stream.fail(domainError) : Stream.never)), ), ), } as unknown as WsRpcProtocolClient; const { activeSession, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(client))); const subscriptionFiber = yield* subscribe( WS_METHODS.subscribeTerminalEvents, {}, { onExpectedFailure: () => Ref.update(expectedFailureCount, (count) => count + 1), retryExpectedFailureAfter: "100 millis", }, ).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.forkChild, ); for (let attempt = 0; attempt < 100; attempt += 1) { if ((yield* Ref.get(expectedFailureCount)) >= 1) { break; } yield* Effect.yieldNow; } expect(yield* Ref.get(subscriptionCount)).toBe(1); expect(yield* Ref.get(expectedFailureCount)).toBe(1); yield* TestClock.adjust("100 millis"); for (let attempt = 0; attempt < 100; attempt += 1) { if ((yield* Ref.get(subscriptionCount)) >= 2) { break; } yield* Effect.yieldNow; } yield* Fiber.interrupt(subscriptionFiber); expect(yield* Ref.get(subscriptionCount)).toBe(2); expect(yield* Ref.get(expectedFailureCount)).toBe(1); }), ); it.effect("does not classify subscription defects as expected failures", () => Effect.gen(function* () { const defect = new Error("subscription invariant failed"); let expectedFailureCount = 0; const client = { [WS_METHODS.subscribeTerminalEvents]: () => Stream.die(defect), } as unknown as WsRpcProtocolClient; const { activeSession, supervisor } = yield* makeHarness(); yield* SubscriptionRef.set(activeSession, Option.some(session(client))); const exit = yield* subscribe( WS_METHODS.subscribeTerminalEvents, {}, { onExpectedFailure: () => Effect.sync(() => { expectedFailureCount += 1; }), }, ).pipe( Stream.runDrain, Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.exit, ); expect(Exit.isFailure(exit)).toBe(true); if (Exit.isFailure(exit)) { expect(Cause.hasDies(exit.cause)).toBe(true); } expect(expectedFailureCount).toBe(0); }), );});
import {
EnvironmentId,
type RelayClientInstallProgressEvent,
WS_METHODS,
} from "@t3tools/contracts";
import { describe, expect, it } from "@effect/vitest";
import * as Cause from "effect/Cause";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
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 { RpcClientError } from "effect/unstable/rpc";
import {
AVAILABLE_CONNECTION_STATE,
PrimaryConnectionTarget,
type PreparedConnection,
type SupervisorConnectionState,
} from "../connection/model.ts";
import * as EnvironmentSupervisor from "../connection/supervisor.ts";
import * as RpcSession from "../rpc/session.ts";
import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
import { EnvironmentRpcRequestObserver, request, runStream, subscribe } from "./client.ts";
const TARGET = new PrimaryConnectionTarget({
environmentId: EnvironmentId.make("environment-1"),
label: "Test environment",
httpBaseUrl: "https://environment.example.test",
wsBaseUrl: "wss://environment.example.test",
});
const INSTALL_CHECKING: RelayClientInstallProgressEvent = {
type: "progress",
stage: "checking",
};
const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = {
type: "progress",
stage: "downloading",
};
function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
return {
client,
initialConfig: Effect.never,
ready: Effect.void,
probe: Effect.void,
closed: Effect.never,
};
}
const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
const state = yield* SubscriptionRef.make<SupervisorConnectionState>(AVAILABLE_CONNECTION_STATE);
const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
Option.none(),
);
const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
const retryCount = yield* Ref.make(0);
const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
target: TARGET,
state,
session: activeSession,
prepared,
connect: Effect.void,
disconnect: Effect.void,
retryNow: Ref.update(retryCount, (count) => count + 1),
} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
return {
activeSession,
retryCount,
supervisor,
};
});
describe("environment RPC", () => {
it.effect("observes unary requests until they complete", () =>
Effect.gen(function* () {
const observations: string[] = [];
const client = {
[WS_METHODS.cloudGetRelayClientStatus]: () =>
Effect.succeed({ status: "available", version: "2026.6.0" }),
} as unknown as WsRpcProtocolClient;
const { activeSession, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
const result = yield* request(WS_METHODS.cloudGetRelayClientStatus, {}).pipe(
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.provideService(
EnvironmentRpcRequestObserver,
EnvironmentRpcRequestObserver.of({
observe: ({ environmentId, method }) =>
Effect.sync(() => {
observations.push(`start:${environmentId}:${method}`);
return Effect.sync(() => {
observations.push(`finish:${environmentId}:${method}`);
});
}),
}),
),
);
expect(result).toEqual({ status: "available", version: "2026.6.0" });
expect(observations).toEqual([
`start:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
`finish:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
]);
}),
);
it.effect("binds finite streaming commands to one active session", () =>
Effect.gen(function* () {
const firstEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
const secondEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
const firstClient = {
[WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(firstEvents),
} as unknown as WsRpcProtocolClient;
const secondClient = {
[WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(secondEvents),
} as unknown as WsRpcProtocolClient;
const { activeSession, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
const resultFiber = yield* runStream(WS_METHODS.cloudInstallRelayClient, {}).pipe(
Stream.take(2),
Stream.runCollect,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.forkChild,
);
yield* Effect.yieldNow;
yield* Queue.offer(firstEvents, INSTALL_CHECKING);
yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
yield* Queue.offer(secondEvents, INSTALL_DOWNLOADING);
yield* Queue.offer(firstEvents, INSTALL_DOWNLOADING);
expect(yield* Fiber.join(resultFiber)).toEqual([INSTALL_CHECKING, INSTALL_DOWNLOADING]);
}),
);
it.effect("switches durable subscriptions when the supervisor replaces the session", () =>
Effect.gen(function* () {
const subscriptions: string[] = [];
const firstClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("first");
return Stream.never;
},
} as unknown as WsRpcProtocolClient;
const secondClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("second");
return Stream.never;
},
} as unknown as WsRpcProtocolClient;
const { activeSession, retryCount, supervisor } = yield* makeHarness();
const awaitSubscriptions = Effect.fn("TestEnvironmentRpc.awaitSubscriptions")(function* (
count: number,
) {
for (let attempt = 0; attempt < 100; attempt += 1) {
if (subscriptions.length >= count) {
return;
}
yield* Effect.yieldNow;
}
return yield* Effect.die(new Error(`Expected ${count} durable subscriptions.`));
});
const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.forkChild,
);
yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
yield* awaitSubscriptions(1);
yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
yield* awaitSubscriptions(2);
yield* Fiber.interrupt(subscriptionFiber);
expect(subscriptions).toEqual(["first", "second"]);
expect(yield* Ref.get(retryCount)).toBe(0);
}),
);
it.effect("keeps durable subscriptions alive across a transport failure and new session", () =>
Effect.gen(function* () {
const subscriptions: string[] = [];
const firstClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("first");
return Stream.fail(
new RpcClientError.RpcClientError({
reason: new RpcClientError.RpcClientDefect({
message: "socket closed",
cause: new Error("socket closed"),
}),
}),
);
},
} as unknown as WsRpcProtocolClient;
const secondClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("second");
return Stream.never;
},
} as unknown as WsRpcProtocolClient;
const { activeSession, retryCount, supervisor } = yield* makeHarness();
const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.forkChild,
);
yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
for (let attempt = 0; attempt < 100 && subscriptions.length < 1; attempt += 1) {
yield* Effect.yieldNow;
}
yield* SubscriptionRef.set(activeSession, Option.none());
yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) {
yield* Effect.yieldNow;
}
yield* Fiber.interrupt(subscriptionFiber);
expect(subscriptions).toEqual(["first", "second"]);
expect(yield* Ref.get(retryCount)).toBe(0);
}),
);
it.effect("surfaces domain subscription failures without reconnecting", () =>
Effect.gen(function* () {
const domainError = new Error("terminal subscription rejected");
const client = {
[WS_METHODS.subscribeTerminalEvents]: () => Stream.fail(domainError),
} as unknown as WsRpcProtocolClient;
const { activeSession, retryCount, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
const error = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.flip,
);
expect(error).toBe(domainError);
expect(yield* Ref.get(retryCount)).toBe(0);
}),
);
it.effect("keeps handled domain failures dormant until a replacement session arrives", () =>
Effect.gen(function* () {
const domainError = new Error("terminal subscription rejected");
const subscriptions: string[] = [];
const observedFailures: Error[] = [];
const firstClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("first");
return Stream.fail(domainError);
},
} as unknown as WsRpcProtocolClient;
const secondClient = {
[WS_METHODS.subscribeTerminalEvents]: () => {
subscriptions.push("second");
return Stream.never;
},
} as unknown as WsRpcProtocolClient;
const { activeSession, retryCount, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
const subscriptionFiber = yield* subscribe(
WS_METHODS.subscribeTerminalEvents,
{},
{
onExpectedFailure: (cause) =>
Effect.sync(() => {
observedFailures.push(Cause.squash(cause) as Error);
}),
},
).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.forkChild,
);
for (let attempt = 0; attempt < 100 && observedFailures.length < 1; attempt += 1) {
yield* Effect.yieldNow;
}
expect(subscriptions).toEqual(["first"]);
expect(observedFailures).toEqual([domainError]);
yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) {
yield* Effect.yieldNow;
}
yield* Fiber.interrupt(subscriptionFiber);
expect(subscriptions).toEqual(["first", "second"]);
expect(yield* Ref.get(retryCount)).toBe(0);
}),
);
it.effect("retries handled domain failures within the same session when configured", () =>
Effect.gen(function* () {
const domainError = new Error("thread not found yet");
const subscriptionCount = yield* Ref.make(0);
const expectedFailureCount = yield* Ref.make(0);
const client = {
[WS_METHODS.subscribeTerminalEvents]: () =>
Stream.unwrap(
Ref.getAndUpdate(subscriptionCount, (count) => count + 1).pipe(
Effect.map((count) => (count === 0 ? Stream.fail(domainError) : Stream.never)),
),
),
} as unknown as WsRpcProtocolClient;
const { activeSession, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
const subscriptionFiber = yield* subscribe(
WS_METHODS.subscribeTerminalEvents,
{},
{
onExpectedFailure: () => Ref.update(expectedFailureCount, (count) => count + 1),
retryExpectedFailureAfter: "100 millis",
},
).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.forkChild,
);
for (let attempt = 0; attempt < 100; attempt += 1) {
if ((yield* Ref.get(expectedFailureCount)) >= 1) {
break;
}
yield* Effect.yieldNow;
}
expect(yield* Ref.get(subscriptionCount)).toBe(1);
expect(yield* Ref.get(expectedFailureCount)).toBe(1);
yield* TestClock.adjust("100 millis");
for (let attempt = 0; attempt < 100; attempt += 1) {
if ((yield* Ref.get(subscriptionCount)) >= 2) {
break;
}
yield* Effect.yieldNow;
}
yield* Fiber.interrupt(subscriptionFiber);
expect(yield* Ref.get(subscriptionCount)).toBe(2);
expect(yield* Ref.get(expectedFailureCount)).toBe(1);
}),
);
it.effect("does not classify subscription defects as expected failures", () =>
Effect.gen(function* () {
const defect = new Error("subscription invariant failed");
let expectedFailureCount = 0;
const client = {
[WS_METHODS.subscribeTerminalEvents]: () => Stream.die(defect),
} as unknown as WsRpcProtocolClient;
const { activeSession, supervisor } = yield* makeHarness();
yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
const exit = yield* subscribe(
WS_METHODS.subscribeTerminalEvents,
{},
{
onExpectedFailure: () =>
Effect.sync(() => {
expectedFailureCount += 1;
}),
},
).pipe(
Stream.runDrain,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.exit,
);
expect(Exit.isFailure(exit)).toBe(true);
if (Exit.isFailure(exit)) {
expect(Cause.hasDies(exit.cause)).toBe(true);
}
expect(expectedFailureCount).toBe(0);
}),
);
});