1import { assert, describe, it } from "@effect/vitest";
2import * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import * as ErrorReporter from "effect/ErrorReporter";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
8import * as Logger from "effect/Logger";
9import * as Queue from "effect/Queue";
10import * as Schema from "effect/Schema";
11import * as Stream from "effect/Stream";
12import { Rpc, RpcClient, RpcGroup, RpcServer } from "effect/rpc";
13
14import { WS_RPC_SERVER_OPTIONS } from "../ws.ts";
15import * as DefectReporter from "./DefectReporter.ts";
16
17class TestRpcs extends RpcGroup.make(
18 Rpc.make("subscribe", { success: Schema.Number, stream: true }),
19 Rpc.make("boom", { success: Schema.Void }),
20) {}
21type TestRpc = RpcGroup.Rpcs<typeof TestRpcs>;
22
23/** Runs `body` with every error log captured. */
24const withErrorLogs = <A, E, R>(
25 body: (logs: Queue.Queue<Cause.Cause<unknown>>) => Effect.Effect<A, E, R>,
26) =>
27 Effect.gen(function* () {
28 const logs = yield* Queue.unbounded<Cause.Cause<unknown>>();
29 const logger = Logger.make(({ cause, logLevel }) => {
30 if (logLevel === "Error") Queue.offerUnsafe(logs, cause);
31 });
32 return yield* body(logs).pipe(
33 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
34 );
35 });
36
37const errorMessage = (cause: Cause.Cause<unknown>) => (Cause.squash(cause) as Error).message;
38
39describe("DefectReporter", () => {
40 it.effect("a dying RPC handler fails alone, and its defect is logged", () =>
41 withErrorLogs((logs) =>
42 Effect.gen(function* () {
43 const subscription = yield* Queue.unbounded<number>();
44 // The same client/server pairing as RpcTest.makeClient, with ws.ts's options.
45 let client!: Effect.Success<
46 ReturnType<typeof RpcClient.makeNoSerialization<TestRpc, never>>
47 >;
48 const server = yield* RpcServer.makeNoSerialization(TestRpcs, {
49 ...WS_RPC_SERVER_OPTIONS,
50 onFromServer: (response) => client.write(response),
51 }).pipe(
52 // Provided to the handlers, as ws.ts does.
53 Effect.provide(
54 TestRpcs.toLayer({
55 subscribe: () => Stream.fromQueue(subscription),
56 boom: () => Effect.die(new Error("handler bug")),
57 }).pipe(Layer.provide(DefectReporter.layer)),
58 ),
59 );
60 client = yield* RpcClient.makeNoSerialization(TestRpcs, {
61 supportsAck: true,
62 onFromClient: ({ message }) => server.write(0, message),
63 });
64
65 const received = yield* Queue.unbounded<number>();
66 const sibling = yield* client.client.subscribe().pipe(
67 Stream.runForEach((value) => Queue.offer(received, value)),
68 Effect.forkScoped,
69 );
70 yield* Queue.offer(subscription, 1);
71 assert.equal(yield* Queue.take(received), 1);
72
73 const boom = yield* Effect.exit(client.client.boom());
74 assert.isTrue(Exit.hasDies(boom));
75 // The server reports before it answers, so the log is already written.
76 const logged = yield* Queue.clear(logs);
77 assert.deepEqual(logged.map(errorMessage), ["handler bug"]);
78
79 // The sibling subscription on the same client keeps delivering.
80 yield* Queue.offer(subscription, 2);
81 const next = yield* Queue.take(received).pipe(
82 Effect.raceFirst(Fiber.await(sibling).pipe(Effect.as("subscription ended"))),
83 );
84 assert.equal(next, 2);
85 assert.equal(yield* Queue.size(logs), 0);
86 }),
87 ).pipe(Effect.scoped),
88 );
89
90 it.effect("logs defects, not typed failures or interrupts", () =>
91 withErrorLogs((logs) =>
92 Effect.gen(function* () {
93 yield* ErrorReporter.report(Cause.fail(new Error("expected")));
94 yield* ErrorReporter.report(Cause.interrupt());
95 yield* ErrorReporter.report(Cause.die(new Error("bug")));
96
97 // Reporters log synchronously, so the logs are already written.
98 assert.deepEqual((yield* Queue.clear(logs)).map(errorMessage), ["bug"]);
99 }).pipe(Effect.provide(DefectReporter.layer)),
100 ),
101 );
102});