1import {
2 DEFAULT_SERVER_SETTINGS,
3 EnvironmentAuthorizationError,
4 EnvironmentId,
5 type RelayClientInstallProgressEvent,
6 type ServerConfigStreamEvent,
7 type ServerLifecycleStreamEvent,
8 WS_METHODS,
9} from "@t3tools/contracts";
10import { describe, expect, it } from "@effect/vitest";
11import * as Cause from "effect/Cause";
12import * as Deferred from "effect/Deferred";
13import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";
15import * as Fiber from "effect/Fiber";
16import * as Option from "effect/Option";
17import * as Queue from "effect/Queue";
18import * as Ref from "effect/Ref";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import * as TestClock from "effect/testing/TestClock";
22import { RpcClientError } from "effect/rpc";
23
24import {
25 AVAILABLE_CONNECTION_STATE,
26 PrimaryConnectionTarget,
27 type PreparedConnection,
28 type SupervisorConnectionState,
29} from "../connection/model.ts";
30import * as EnvironmentSupervisor from "../connection/supervisor.ts";
31import * as RpcSession from "../rpc/session.ts";
32import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
33import {
34 EnvironmentRpcRequestObserver,
35 request,
36 requestGuarded,
37 runStream,
38 subscribe,
39 subscribeDynamicWithSession,
40} from "./client.ts";
41
42const TARGET = new PrimaryConnectionTarget({
43 environmentId: EnvironmentId.make("environment-1"),
44 label: "Test environment",
45 httpBaseUrl: "https://environment.example.test",
46 wsBaseUrl: "wss://environment.example.test",
47});
48
49const INSTALL_CHECKING: RelayClientInstallProgressEvent = {
50 type: "progress",
51 stage: "checking",
52};
53const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = {
54 type: "progress",
55 stage: "downloading",
56};
57
58function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
59 return {
60 client,
61 initialConfig: Effect.never,
62 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
63 ready: Effect.void,
64 probe: Effect.void,
65 closed: Effect.never,
66 };
67}
68
69const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
70 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(AVAILABLE_CONNECTION_STATE);
71 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
72 Option.none(),
73 );
74 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
75 const retryCount = yield* Ref.make(0);
76 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
77 target: TARGET,
78 state,
79 session: activeSession,
80 prepared,
81 connect: Effect.void,
82 disconnect: Effect.void,
83 retryNow: Ref.update(retryCount, (count) => count + 1),
84 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
85 return {
86 activeSession,
87 retryCount,
88 supervisor,
89 };
90});
91
92describe("environment RPC", () => {
93 it.effect("reads managed workspace status without a command boundary and denies creation", () =>
94 Effect.gen(function* () {
95 let creates = 0;
96 const client = {
97 [WS_METHODS.managedWorkspacesStatus]: () => Effect.succeed({ projectId: null }),
98 [WS_METHODS.managedWorkspacesCreate]: () => {
99 creates += 1;
100 return Effect.die("Protected workspace creation reached the transport");
101 },
102 } as unknown as WsRpcProtocolClient;
103 const { activeSession, supervisor } = yield* makeHarness();
104 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
105
106 const status = yield* request(WS_METHODS.managedWorkspacesStatus, {}).pipe(
107 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
108 );
109 expect(status).toEqual({ projectId: null });
110
111 const create = yield* requestGuarded(WS_METHODS.managedWorkspacesCreate, {}).pipe(
112 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
113 Effect.flip,
114 );
115 expect(create._tag).toBe("EnvironmentAuthorizationError");
116 expect(creates).toBe(0);
117 }),
118 );
119
120 it.effect("reuses the session config stream instead of opening a duplicate subscription", () =>
121 Effect.gen(function* () {
122 const event: ServerConfigStreamEvent = {
123 version: 1,
124 type: "settingsUpdated",
125 payload: { settings: DEFAULT_SERVER_SETTINGS },
126 };
127 let duplicateSubscriptions = 0;
128 const client = {
129 [WS_METHODS.subscribeServerConfig]: () => {
130 duplicateSubscriptions += 1;
131 return Stream.never;
132 },
133 } as unknown as WsRpcProtocolClient;
134 const { activeSession, supervisor } = yield* makeHarness();
135 yield* SubscriptionRef.set(
136 activeSession,
137 Option.some({
138 ...session(client),
139 subscribeServerConfig: () => Stream.succeed(event),
140 }),
141 );
142
143 const received = yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe(
144 Stream.runHead,
145 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
146 );
147
148 expect(received).toEqual(Option.some(event));
149 expect(duplicateSubscriptions).toBe(0);
150 }),
151 );
152
153 it.effect("observes unary requests until they complete", () =>
154 Effect.gen(function* () {
155 const observations: string[] = [];
156 const client = {
157 [WS_METHODS.cloudGetRelayClientStatus]: () =>
158 Effect.succeed({ status: "available", version: "2026.6.0" }),
159 } as unknown as WsRpcProtocolClient;
160 const { activeSession, supervisor } = yield* makeHarness();
161 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
162
163 const result = yield* request(WS_METHODS.cloudGetRelayClientStatus, {}).pipe(
164 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
165 Effect.provideService(
166 EnvironmentRpcRequestObserver,
167 EnvironmentRpcRequestObserver.of({
168 observe: ({ environmentId, method }) =>
169 Effect.sync(() => {
170 observations.push(`start:${environmentId}:${method}`);
171 return Effect.sync(() => {
172 observations.push(`finish:${environmentId}:${method}`);
173 });
174 }),
175 }),
176 ),
177 );
178
179 expect(result).toEqual({ status: "available", version: "2026.6.0" });
180 expect(observations).toEqual([
181 `start:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
182 `finish:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
183 ]);
184 }),
185 );
186
187 it.effect("binds finite streaming commands to one active session", () =>
188 Effect.gen(function* () {
189 const firstEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
190 const secondEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
191 const firstClient = {
192 [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(firstEvents),
193 } as unknown as WsRpcProtocolClient;
194 const secondClient = {
195 [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(secondEvents),
196 } as unknown as WsRpcProtocolClient;
197 const { activeSession, supervisor } = yield* makeHarness();
198
199 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
200 const resultFiber = yield* runStream(WS_METHODS.cloudInstallRelayClient, {}).pipe(
201 Stream.take(2),
202 Stream.runCollect,
203 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
204 Effect.forkChild,
205 );
206 yield* Effect.yieldNow;
207
208 yield* Queue.offer(firstEvents, INSTALL_CHECKING);
209 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
210 yield* Queue.offer(secondEvents, INSTALL_DOWNLOADING);
211 yield* Queue.offer(firstEvents, INSTALL_DOWNLOADING);
212
213 expect(yield* Fiber.join(resultFiber)).toEqual([INSTALL_CHECKING, INSTALL_DOWNLOADING]);
214 }),
215 );
216
217 it.effect("switches durable subscriptions when the supervisor replaces the session", () =>
218 Effect.gen(function* () {
219 const subscriptions: string[] = [];
220 const firstClient = {
221 [WS_METHODS.subscribeTerminalEvents]: () => {
222 subscriptions.push("first");
223 return Stream.never;
224 },
225 } as unknown as WsRpcProtocolClient;
226 const secondClient = {
227 [WS_METHODS.subscribeTerminalEvents]: () => {
228 subscriptions.push("second");
229 return Stream.never;
230 },
231 } as unknown as WsRpcProtocolClient;
232 const { activeSession, retryCount, supervisor } = yield* makeHarness();
233 const awaitSubscriptions = Effect.fn("TestEnvironmentRpc.awaitSubscriptions")(function* (
234 count: number,
235 ) {
236 for (let attempt = 0; attempt < 100; attempt += 1) {
237 if (subscriptions.length >= count) {
238 return;
239 }
240 yield* Effect.yieldNow;
241 }
242 return yield* Effect.die(new Error(`Expected ${count} durable subscriptions.`));
243 });
244
245 const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
246 Stream.runDrain,
247 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
248 Effect.forkChild,
249 );
250 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
251 yield* awaitSubscriptions(1);
252 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
253 yield* awaitSubscriptions(2);
254 yield* Fiber.interrupt(subscriptionFiber);
255
256 expect(subscriptions).toEqual(["first", "second"]);
257 expect(yield* Ref.get(retryCount)).toBe(0);
258 }),
259 );
260
261 it.effect("keeps the producer session on an old value buffered across a session switch", () =>
262 Effect.gen(function* () {
263 const firstSubscribed = yield* Deferred.make<void>();
264 const secondSubscribed = yield* Deferred.make<void>();
265 const firstValueBlocked = yield* Deferred.make<void>();
266 const releaseFirstValue = yield* Deferred.make<void>();
267 const firstValue = { source: "first", index: 1 } as unknown as ServerLifecycleStreamEvent;
268 const bufferedFirstValue = {
269 source: "first",
270 index: 2,
271 } as unknown as ServerLifecycleStreamEvent;
272 const secondValue = { source: "second", index: 1 } as unknown as ServerLifecycleStreamEvent;
273 const firstClient = {
274 [WS_METHODS.subscribeServerLifecycle]: () =>
275 Stream.fromEffect(Deferred.succeed(firstSubscribed, undefined)).pipe(
276 Stream.drain,
277 Stream.concat(Stream.fromIterable([firstValue, bufferedFirstValue])),
278 Stream.concat(Stream.never),
279 ),
280 } as unknown as WsRpcProtocolClient;
281 const secondClient = {
282 [WS_METHODS.subscribeServerLifecycle]: () =>
283 Stream.fromEffect(Deferred.succeed(secondSubscribed, undefined)).pipe(
284 Stream.drain,
285 Stream.concat(Stream.make(secondValue)),
286 Stream.concat(Stream.never),
287 ),
288 } as unknown as WsRpcProtocolClient;
289 const firstSession = session(firstClient);
290 const secondSession = session(secondClient);
291 const { activeSession, supervisor } = yield* makeHarness();
292
293 const resultFiber = yield* subscribeDynamicWithSession(
294 WS_METHODS.subscribeServerLifecycle,
295 () => Effect.succeed({}),
296 ).pipe(
297 Stream.mapEffect(([producerSession, value]) =>
298 value === firstValue
299 ? Deferred.succeed(firstValueBlocked, undefined).pipe(
300 Effect.andThen(Deferred.await(releaseFirstValue)),
301 Effect.as([producerSession, value] as const),
302 )
303 : Effect.succeed([producerSession, value] as const),
304 ),
305 Stream.take(3),
306 Stream.runCollect,
307 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
308 Effect.forkChild,
309 );
310
311 yield* SubscriptionRef.set(activeSession, Option.some(firstSession));
312 yield* Deferred.await(firstSubscribed);
313 yield* Deferred.await(firstValueBlocked);
314 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
315 yield* Deferred.await(secondSubscribed);
316 yield* Deferred.succeed(releaseFirstValue, undefined);
317
318 const result = yield* Fiber.join(resultFiber);
319 expect(result).toEqual([
320 [firstSession, firstValue],
321 [firstSession, bufferedFirstValue],
322 [secondSession, secondValue],
323 ]);
324 }),
325 );
326
327 it.effect("keeps durable subscriptions alive across a transport failure and new session", () =>
328 Effect.gen(function* () {
329 const subscriptions: string[] = [];
330 const firstClient = {
331 [WS_METHODS.subscribeTerminalEvents]: () => {
332 subscriptions.push("first");
333 return Stream.fail(
334 new RpcClientError.RpcClientError({
335 reason: new RpcClientError.RpcClientDefect({
336 message: "socket closed",
337 cause: new Error("socket closed"),
338 }),
339 }),
340 );
341 },
342 } as unknown as WsRpcProtocolClient;
343 const secondClient = {
344 [WS_METHODS.subscribeTerminalEvents]: () => {
345 subscriptions.push("second");
346 return Stream.never;
347 },
348 } as unknown as WsRpcProtocolClient;
349 const { activeSession, retryCount, supervisor } = yield* makeHarness();
350
351 const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
352 Stream.runDrain,
353 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
354 Effect.forkChild,
355 );
356 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
357 for (let attempt = 0; attempt < 100 && subscriptions.length < 1; attempt += 1) {
358 yield* Effect.yieldNow;
359 }
360 yield* SubscriptionRef.set(activeSession, Option.none());
361 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
362
363 for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) {
364 yield* Effect.yieldNow;
365 }
366 yield* Fiber.interrupt(subscriptionFiber);
367
368 expect(subscriptions).toEqual(["first", "second"]);
369 expect(yield* Ref.get(retryCount)).toBe(0);
370 }),
371 );
372
373 it.effect("surfaces domain subscription failures without reconnecting", () =>
374 Effect.gen(function* () {
375 const domainError = new Error("terminal subscription rejected");
376 const client = {
377 [WS_METHODS.subscribeTerminalEvents]: () => Stream.fail(domainError),
378 } as unknown as WsRpcProtocolClient;
379 const { activeSession, retryCount, supervisor } = yield* makeHarness();
380
381 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
382 const error = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe(
383 Stream.runDrain,
384 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
385 Effect.flip,
386 );
387
388 expect(error).toBe(domainError);
389 expect(yield* Ref.get(retryCount)).toBe(0);
390 }),
391 );
392
393 it.effect("keeps handled domain failures dormant until a replacement session arrives", () =>
394 Effect.gen(function* () {
395 const domainError = new Error("terminal subscription rejected");
396 const subscriptions: string[] = [];
397 const observedFailures: Error[] = [];
398 const firstClient = {
399 [WS_METHODS.subscribeTerminalEvents]: () => {
400 subscriptions.push("first");
401 return Stream.fail(domainError);
402 },
403 } as unknown as WsRpcProtocolClient;
404 const secondClient = {
405 [WS_METHODS.subscribeTerminalEvents]: () => {
406 subscriptions.push("second");
407 return Stream.never;
408 },
409 } as unknown as WsRpcProtocolClient;
410 const { activeSession, retryCount, supervisor } = yield* makeHarness();
411
412 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
413 const subscriptionFiber = yield* subscribe(
414 WS_METHODS.subscribeTerminalEvents,
415 {},
416 {
417 onExpectedFailure: (cause) =>
418 Effect.sync(() => {
419 observedFailures.push(Cause.squash(cause) as Error);
420 }),
421 },
422 ).pipe(
423 Stream.runDrain,
424 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
425 Effect.forkChild,
426 );
427 for (let attempt = 0; attempt < 100 && observedFailures.length < 1; attempt += 1) {
428 yield* Effect.yieldNow;
429 }
430
431 expect(subscriptions).toEqual(["first"]);
432 expect(observedFailures).toEqual([domainError]);
433
434 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
435 for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) {
436 yield* Effect.yieldNow;
437 }
438 yield* Fiber.interrupt(subscriptionFiber);
439
440 expect(subscriptions).toEqual(["first", "second"]);
441 expect(yield* Ref.get(retryCount)).toBe(0);
442 }),
443 );
444
445 it.effect("retries handled domain failures within the same session when configured", () =>
446 Effect.gen(function* () {
447 const domainError = new Error("thread not found yet");
448 const subscriptionCount = yield* Ref.make(0);
449 const expectedFailureCount = yield* Ref.make(0);
450 const client = {
451 [WS_METHODS.subscribeTerminalEvents]: () =>
452 Stream.unwrap(
453 Ref.getAndUpdate(subscriptionCount, (count) => count + 1).pipe(
454 Effect.map((count) => (count === 0 ? Stream.fail(domainError) : Stream.never)),
455 ),
456 ),
457 } as unknown as WsRpcProtocolClient;
458 const { activeSession, supervisor } = yield* makeHarness();
459
460 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
461 const subscriptionFiber = yield* subscribe(
462 WS_METHODS.subscribeTerminalEvents,
463 {},
464 {
465 onExpectedFailure: () => Ref.update(expectedFailureCount, (count) => count + 1),
466 retryExpectedFailureAfter: "100 millis",
467 },
468 ).pipe(
469 Stream.runDrain,
470 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
471 Effect.forkChild,
472 );
473 for (let attempt = 0; attempt < 100; attempt += 1) {
474 if ((yield* Ref.get(expectedFailureCount)) >= 1) {
475 break;
476 }
477 yield* Effect.yieldNow;
478 }
479
480 expect(yield* Ref.get(subscriptionCount)).toBe(1);
481 expect(yield* Ref.get(expectedFailureCount)).toBe(1);
482
483 yield* TestClock.adjust("100 millis");
484 for (let attempt = 0; attempt < 100; attempt += 1) {
485 if ((yield* Ref.get(subscriptionCount)) >= 2) {
486 break;
487 }
488 yield* Effect.yieldNow;
489 }
490 yield* Fiber.interrupt(subscriptionFiber);
491
492 expect(yield* Ref.get(subscriptionCount)).toBe(2);
493 expect(yield* Ref.get(expectedFailureCount)).toBe(1);
494 }),
495 );
496
497 it.effect("doubles the retry delay for repeated failures and resets it after a value", () =>
498 Effect.gen(function* () {
499 const domainError = new Error("thread not hydrated yet");
500 const subscriptions = yield* Queue.unbounded<number>();
501 const failures = yield* Queue.unbounded<void>();
502 let attempts = 0;
503 const client = {
504 [WS_METHODS.subscribeTerminalEvents]: () =>
505 Stream.unwrap(
506 Effect.sync(() => {
507 attempts += 1;
508 return attempts;
509 }).pipe(
510 Effect.tap((attempt) => Queue.offer(subscriptions, attempt)),
511 Effect.map((attempt) => {
512 if (attempt <= 2) return Stream.fail(domainError);
513 if (attempt === 3)
514 return Stream.concat(Stream.make("event"), Stream.fail(domainError));
515 return Stream.never;
516 }),
517 ),
518 ),
519 } as unknown as WsRpcProtocolClient;
520 const { activeSession, supervisor } = yield* makeHarness();
521
522 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
523 const subscriptionFiber = yield* subscribe(
524 WS_METHODS.subscribeTerminalEvents,
525 {},
526 {
527 onExpectedFailure: () => Queue.offer(failures, undefined),
528 retryExpectedFailureAfter: "100 millis",
529 },
530 ).pipe(
531 Stream.runDrain,
532 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
533 Effect.forkChild,
534 );
535
536 expect(yield* Queue.take(subscriptions)).toBe(1);
537 yield* Queue.take(failures);
538 yield* TestClock.adjust("100 millis");
539 expect(yield* Queue.take(subscriptions)).toBe(2);
540 yield* Queue.take(failures);
541 // The second retry waits 200ms, so 100ms is not enough.
542 yield* TestClock.adjust("100 millis");
543 expect(Option.isNone(yield* Queue.poll(subscriptions))).toBe(true);
544 yield* TestClock.adjust("100 millis");
545 expect(yield* Queue.take(subscriptions)).toBe(3);
546 // Attempt 3 delivered a value before failing, so the delay is back to 100ms.
547 yield* Queue.take(failures);
548 yield* TestClock.adjust("100 millis");
549 expect(yield* Queue.take(subscriptions)).toBe(4);
550 yield* Fiber.interrupt(subscriptionFiber);
551 }),
552 );
553
554 it.effect("waits for the next session after an authorization failure", () =>
555 Effect.gen(function* () {
556 const subscriptions = yield* Queue.unbounded<void>();
557 const failed = yield* Deferred.make<void>();
558 let attempts = 0;
559 const client = {
560 [WS_METHODS.subscribeTerminalEvents]: () =>
561 Stream.unwrap(
562 Queue.offer(subscriptions, undefined).pipe(
563 Effect.map(() => {
564 attempts += 1;
565 return attempts === 1
566 ? Stream.fail(
567 new EnvironmentAuthorizationError({
568 message: "Missing scope",
569 requiredScope: "orchestration:read",
570 }),
571 )
572 : Stream.never;
573 }),
574 ),
575 ),
576 } as unknown as WsRpcProtocolClient;
577 const { activeSession, supervisor } = yield* makeHarness();
578
579 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
580 const subscriptionFiber = yield* subscribe(
581 WS_METHODS.subscribeTerminalEvents,
582 {},
583 {
584 onExpectedFailure: () => Deferred.succeed(failed, undefined).pipe(Effect.asVoid),
585 retryExpectedFailureAfter: "100 millis",
586 },
587 ).pipe(
588 Stream.runDrain,
589 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
590 Effect.forkChild,
591 );
592
593 yield* Queue.take(subscriptions);
594 yield* Deferred.await(failed);
595 yield* TestClock.adjust("1 minute");
596 expect(Option.isNone(yield* Queue.poll(subscriptions))).toBe(true);
597
598 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
599 yield* Queue.take(subscriptions);
600 yield* Fiber.interrupt(subscriptionFiber);
601 expect(attempts).toBe(2);
602 }),
603 );
604
605 it.effect.each(["input", "stream"] as const)(
606 "does not classify %s subscription defects as expected failures",
607 (where) =>
608 Effect.gen(function* () {
609 const defect = new Error("subscription invariant failed");
610 let expectedFailureCount = 0;
611 let inputs = 0;
612 let streams = 0;
613 const observedDefects: unknown[] = [];
614 const client = {
615 [WS_METHODS.subscribeTerminalEvents]: () => {
616 streams += 1;
617 return where === "stream" ? Stream.die(defect) : Stream.never;
618 },
619 } as unknown as WsRpcProtocolClient;
620 const { activeSession, supervisor } = yield* makeHarness();
621
622 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
623 const exit = yield* subscribeDynamicWithSession(
624 WS_METHODS.subscribeTerminalEvents,
625 () =>
626 Effect.sync(() => {
627 inputs += 1;
628 }).pipe(Effect.andThen(where === "input" ? Effect.die(defect) : Effect.succeed({}))),
629 {
630 onDefect: (cause) =>
631 Effect.sync(() => {
632 observedDefects.push(Cause.squash(cause));
633 }),
634 onExpectedFailure: () =>
635 Effect.sync(() => {
636 expectedFailureCount += 1;
637 }),
638 retryExpectedFailureAfter: "250 millis",
639 },
640 ).pipe(
641 Stream.runDrain,
642 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
643 Effect.exit,
644 );
645
646 expect(Exit.isFailure(exit)).toBe(true);
647 if (Exit.isFailure(exit)) {
648 expect(Cause.hasDies(exit.cause)).toBe(true);
649 expect(Cause.squash(exit.cause)).toBe(defect);
650 }
651 expect(inputs).toBe(1);
652 expect(streams).toBe(where === "input" ? 0 : 1);
653 expect(expectedFailureCount).toBe(0);
654 expect(observedDefects).toEqual([defect]);
655 }),
656 );
657
658 it.effect("reports an initializer defect once after an expected failure retries", () =>
659 Effect.gen(function* () {
660 const defect = new Error("Synthetic retry initializer defect");
661 const expectedFailure = yield* Deferred.make<void>();
662 const observations: string[] = [];
663 const observedDefects: unknown[] = [];
664 let inputs = 0;
665 const client = {
666 [WS_METHODS.subscribeTerminalEvents]: () => {
667 observations.push("stream");
668 return Stream.fail(new Error("subscription not ready"));
669 },
670 } as unknown as WsRpcProtocolClient;
671 const { activeSession, supervisor } = yield* makeHarness();
672 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
673 const fiber = yield* subscribeDynamicWithSession(
674 WS_METHODS.subscribeTerminalEvents,
675 () =>
676 Effect.sync(() => {
677 inputs += 1;
678 observations.push(`input ${inputs}`);
679 return inputs;
680 }).pipe(
681 Effect.flatMap((attempt) => (attempt === 1 ? Effect.succeed({}) : Effect.die(defect))),
682 ),
683 {
684 onDefect: (cause) =>
685 Effect.sync(() => {
686 observations.push("defect");
687 observedDefects.push(Cause.squash(cause));
688 }),
689 onExpectedFailure: () =>
690 Effect.sync(() => {
691 observations.push("expected failure");
692 }).pipe(Effect.andThen(Deferred.succeed(expectedFailure, undefined)), Effect.asVoid),
693 retryExpectedFailureAfter: "250 millis",
694 },
695 ).pipe(
696 Stream.runDrain,
697 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
698 Effect.exit,
699 Effect.forkChild,
700 );
701 yield* Deferred.await(expectedFailure);
702 yield* TestClock.adjust("250 millis");
703 const exit = yield* Fiber.join(fiber);
704 expect(Exit.isFailure(exit)).toBe(true);
705 if (Exit.isFailure(exit)) {
706 expect(Cause.hasDies(exit.cause)).toBe(true);
707 expect(Cause.squash(exit.cause)).toBe(defect);
708 }
709 expect(observations).toEqual(["input 1", "stream", "expected failure", "input 2", "defect"]);
710 expect(observedDefects).toEqual([defect]);
711 }),
712 );
713});