1import {
2 RpcDecodeError,
3 RpcCallError,
4 RpcRemoteStreamError,
5 encodeRpcError,
6 decodeRpcResult,
7 decodeRpcValue,
8 ErrorTag,
9 StreamTag,
10 StreamErrorTag,
11 isRpcErrorEnvelope,
12 isRpcStreamEnvelope,
13 fromRpcReadableStream,
14 fromRpcStreamEnvelope,
15 toRpcStream,
16 makeRpcStub,
17 type RpcErrorEnvelope,
18 type RpcStreamEnvelope,
19} from "@/Cloudflare/Workers/Rpc";
20import { describe, expect, it } from "alchemy-test";
21import * as Cause from "effect/Cause";
22import * as Data from "effect/Data";
23import * as Effect from "effect/Effect";
24import * as Exit from "effect/Exit";
25import * as Stream from "effect/Stream";
26
27class MyError extends Data.TaggedError("MyError")<{
28 readonly message: string;
29}> {}
30
31// ---------------------------------------------------------------------------
32// RpcDecodeError
33// ---------------------------------------------------------------------------
34
35describe(
36 "RpcDecodeError",
37 {
38 tags: [
39 "unit",
40 "provider:cloudflare",
41 "provider:cloudflare:worker",
42 "local",
43 ],
44 },
45 () => {
46 it.effect("message delegates to Error.message when cause is an Error", () =>
47 Effect.gen(function* () {
48 const err = new RpcDecodeError({ cause: new Error("inner") });
49 expect(err._tag).toBe("RpcDecodeError");
50 expect(err.message).toBe("inner");
51 }),
52 );
53
54 it.effect("message stringifies non-Error cause", () =>
55 Effect.gen(function* () {
56 const err = new RpcDecodeError({ cause: 42 });
57 expect(err.message).toBe("42");
58 }),
59 );
60 },
61);
62
63// ---------------------------------------------------------------------------
64// RpcCallError
65// ---------------------------------------------------------------------------
66
67describe(
68 "RpcCallError",
69 {
70 tags: [
71 "unit",
72 "provider:cloudflare",
73 "provider:cloudflare:worker",
74 "local",
75 ],
76 },
77 () => {
78 it.effect("message includes method name and Error.message", () =>
79 Effect.gen(function* () {
80 const err = new RpcCallError({
81 method: "doStuff",
82 cause: new Error("boom"),
83 });
84 expect(err._tag).toBe("RpcCallError");
85 expect(err.message).toBe('RPC call to "doStuff" failed: boom');
86 }),
87 );
88
89 it.effect("message stringifies non-Error cause", () =>
90 Effect.gen(function* () {
91 const err = new RpcCallError({ method: "doStuff", cause: "oops" });
92 expect(err.message).toBe('RPC call to "doStuff" failed: oops');
93 }),
94 );
95 },
96);
97
98// ---------------------------------------------------------------------------
99// isRpcErrorEnvelope
100// ---------------------------------------------------------------------------
101
102describe(
103 "isRpcErrorEnvelope",
104 {
105 tags: [
106 "unit",
107 "provider:cloudflare",
108 "provider:cloudflare:worker",
109 "local",
110 ],
111 },
112 () => {
113 it.effect("detects valid envelope", () =>
114 Effect.gen(function* () {
115 const envelope: RpcErrorEnvelope = {
116 _tag: ErrorTag,
117 error: { message: "boom" },
118 };
119 expect(isRpcErrorEnvelope(envelope)).toBe(true);
120 }),
121 );
122
123 it.effect("rejects non-envelope values", () =>
124 Effect.gen(function* () {
125 expect(isRpcErrorEnvelope(null)).toBe(false);
126 expect(isRpcErrorEnvelope(undefined)).toBe(false);
127 expect(isRpcErrorEnvelope(42)).toBe(false);
128 expect(isRpcErrorEnvelope("hello")).toBe(false);
129 expect(isRpcErrorEnvelope({ _tag: "Other" })).toBe(false);
130 expect(isRpcErrorEnvelope({ _tag: ErrorTag })).toBe(false);
131 }),
132 );
133 },
134);
135
136// ---------------------------------------------------------------------------
137// isRpcStreamEnvelope
138// ---------------------------------------------------------------------------
139
140describe(
141 "isRpcStreamEnvelope",
142 {
143 tags: [
144 "unit",
145 "provider:cloudflare",
146 "provider:cloudflare:worker",
147 "local",
148 ],
149 },
150 () => {
151 it.effect("detects valid bytes envelope", () =>
152 Effect.gen(function* () {
153 const envelope: RpcStreamEnvelope = {
154 _tag: StreamTag,
155 encoding: "bytes",
156 body: new ReadableStream(),
157 };
158 expect(isRpcStreamEnvelope(envelope)).toBe(true);
159 }),
160 );
161
162 it.effect("detects valid jsonl envelope", () =>
163 Effect.gen(function* () {
164 const envelope: RpcStreamEnvelope = {
165 _tag: StreamTag,
166 encoding: "jsonl",
167 body: new ReadableStream(),
168 };
169 expect(isRpcStreamEnvelope(envelope)).toBe(true);
170 }),
171 );
172
173 it.effect("rejects missing or wrong fields", () =>
174 Effect.gen(function* () {
175 expect(isRpcStreamEnvelope(null)).toBe(false);
176 expect(isRpcStreamEnvelope(42)).toBe(false);
177 expect(isRpcStreamEnvelope({ _tag: StreamTag })).toBe(false);
178 expect(
179 isRpcStreamEnvelope({
180 _tag: StreamTag,
181 encoding: "xml",
182 body: new ReadableStream(),
183 }),
184 ).toBe(false);
185 expect(
186 isRpcStreamEnvelope({
187 _tag: StreamTag,
188 encoding: "jsonl",
189 body: "not a stream",
190 }),
191 ).toBe(false);
192 }),
193 );
194 },
195);
196
197// ---------------------------------------------------------------------------
198// encodeRpcError
199// ---------------------------------------------------------------------------
200
201describe(
202 "encodeRpcError",
203 {
204 tags: [
205 "unit",
206 "provider:cloudflare",
207 "provider:cloudflare:worker",
208 "local",
209 ],
210 },
211 () => {
212 it.effect("preserves tagged error fields", () =>
213 Effect.gen(function* () {
214 const error = new MyError({ message: "BOOF" });
215 const encoded = encodeRpcError(error) as Record<string, unknown>;
216 expect(encoded._tag).toBe("MyError");
217 expect(encoded.message).toBe("BOOF");
218 }),
219 );
220
221 it.effect("normalizes plain Error to name/message/stack", () =>
222 Effect.gen(function* () {
223 const error = new Error("plain failure");
224 const encoded = encodeRpcError(error) as Record<string, unknown>;
225 expect(encoded.name).toBe("Error");
226 expect(encoded.message).toBe("plain failure");
227 expect(encoded.stack).toBeDefined();
228 }),
229 );
230
231 it.effect("passes through primitives", () =>
232 Effect.gen(function* () {
233 expect(encodeRpcError("string error")).toBe("string error");
234 expect(encodeRpcError(42)).toBe(42);
235 expect(encodeRpcError(null)).toBe(null);
236 expect(encodeRpcError(undefined)).toBe(undefined);
237 }),
238 );
239
240 it.effect("passes through plain objects", () =>
241 Effect.gen(function* () {
242 const obj = { code: 404, detail: "not found" };
243 expect(encodeRpcError(obj)).toBe(obj);
244 }),
245 );
246 },
247);
248
249// ---------------------------------------------------------------------------
250// Helper: create a ReadableStream from a string
251// ---------------------------------------------------------------------------
252
253const textToReadableStream = (text: string): ReadableStream<Uint8Array> => {
254 const encoder = new TextEncoder();
255 return new ReadableStream({
256 start(controller) {
257 controller.enqueue(encoder.encode(text));
258 controller.close();
259 },
260 });
261};
262
263const bytesToReadableStream = (
264 chunks: Uint8Array[],
265): ReadableStream<Uint8Array> =>
266 new ReadableStream({
267 start(controller) {
268 for (const chunk of chunks) {
269 controller.enqueue(chunk);
270 }
271 controller.close();
272 },
273 });
274
275// ---------------------------------------------------------------------------
276// fromRpcReadableStream
277// ---------------------------------------------------------------------------
278
279describe(
280 "fromRpcReadableStream",
281 {
282 tags: [
283 "unit",
284 "provider:cloudflare",
285 "provider:cloudflare:worker",
286 "local",
287 ],
288 },
289 () => {
290 it.effect("bytes encoding passes raw Uint8Array chunks through", () =>
291 Effect.gen(function* () {
292 const data = new TextEncoder().encode("hello");
293 const body = bytesToReadableStream([data]);
294 const stream = fromRpcReadableStream(body, "bytes");
295 const chunks = yield* Stream.runCollect(stream);
296 expect(chunks).toEqual([data]);
297 }),
298 );
299
300 it.effect("jsonl encoding parses JSON lines", () =>
301 Effect.gen(function* () {
302 const body = textToReadableStream('{"a":1}\n{"b":2}\n');
303 const stream = fromRpcReadableStream(body, "jsonl");
304 const chunks = yield* Stream.runCollect(stream);
305 expect(chunks).toEqual([{ a: 1 }, { b: 2 }]);
306 }),
307 );
308
309 it.effect("jsonl encoding produces RpcDecodeError on malformed JSON", () =>
310 Effect.gen(function* () {
311 const body = textToReadableStream("not json\n");
312 const stream = fromRpcReadableStream(body, "jsonl");
313 const exit = yield* Effect.exit(Stream.runCollect(stream));
314 expect(Exit.isFailure(exit)).toBe(true);
315 }),
316 );
317
318 it.effect("jsonl encoding skips empty lines", () =>
319 Effect.gen(function* () {
320 const body = textToReadableStream('{"x":1}\n\n\n{"y":2}\n');
321 const stream = fromRpcReadableStream(body, "jsonl");
322 const chunks = yield* Stream.runCollect(stream);
323 expect(chunks).toEqual([{ x: 1 }, { y: 2 }]);
324 }),
325 );
326 },
327);
328
329// ---------------------------------------------------------------------------
330// fromRpcStreamEnvelope
331// ---------------------------------------------------------------------------
332
333describe(
334 "fromRpcStreamEnvelope",
335 {
336 tags: [
337 "unit",
338 "provider:cloudflare",
339 "provider:cloudflare:worker",
340 "local",
341 ],
342 },
343 () => {
344 it.effect("delegates to fromRpcReadableStream", () =>
345 Effect.gen(function* () {
346 const body = textToReadableStream('{"v":99}\n');
347 const envelope: RpcStreamEnvelope = {
348 _tag: StreamTag,
349 encoding: "jsonl",
350 body,
351 };
352 const chunks = yield* Stream.runCollect(
353 fromRpcStreamEnvelope(envelope),
354 );
355 expect(chunks).toEqual([{ v: 99 }]);
356 }),
357 );
358 },
359);
360
361// ---------------------------------------------------------------------------
362// decodeRpcValue
363// ---------------------------------------------------------------------------
364
365describe(
366 "decodeRpcValue",
367 {
368 tags: [
369 "unit",
370 "provider:cloudflare",
371 "provider:cloudflare:worker",
372 "local",
373 ],
374 },
375 () => {
376 it.effect("passes through plain values", () =>
377 Effect.gen(function* () {
378 expect(decodeRpcValue("hello")).toBe("hello");
379 expect(decodeRpcValue(42)).toBe(42);
380 expect(decodeRpcValue(null)).toBe(null);
381 }),
382 );
383
384 it.effect("converts stream envelope to Effect Stream", () =>
385 Effect.gen(function* () {
386 const body = textToReadableStream('{"k":1}\n');
387 const envelope: RpcStreamEnvelope = {
388 _tag: StreamTag,
389 encoding: "jsonl",
390 body,
391 };
392 const result = decodeRpcValue(envelope);
393 expect(Stream.isStream(result)).toBe(true);
394 const chunks = yield* Stream.runCollect(result as Stream.Stream<any>);
395 expect(chunks).toEqual([{ k: 1 }]);
396 }),
397 );
398
399 it.effect("converts bare ReadableStream to bytes Effect Stream", () =>
400 Effect.gen(function* () {
401 const data = new TextEncoder().encode("raw");
402 const body = bytesToReadableStream([data]);
403 const result = decodeRpcValue(body);
404 expect(Stream.isStream(result)).toBe(true);
405 const chunks = yield* Stream.runCollect(result as Stream.Stream<any>);
406 expect(chunks).toEqual([data]);
407 }),
408 );
409 },
410);
411
412// ---------------------------------------------------------------------------
413// decodeRpcResult
414// ---------------------------------------------------------------------------
415
416describe(
417 "decodeRpcResult",
418 {
419 tags: [
420 "unit",
421 "provider:cloudflare",
422 "provider:cloudflare:worker",
423 "local",
424 ],
425 },
426 () => {
427 it.effect("succeeds for plain values", () =>
428 Effect.gen(function* () {
429 const result = yield* decodeRpcResult("hello");
430 expect(result).toBe("hello");
431 }),
432 );
433
434 it.effect("succeeds for numeric values", () =>
435 Effect.gen(function* () {
436 const result = yield* decodeRpcResult(42);
437 expect(result).toBe(42);
438 }),
439 );
440
441 it.effect("fails for error envelopes with tagged error", () =>
442 Effect.gen(function* () {
443 const envelope: RpcErrorEnvelope = {
444 _tag: ErrorTag,
445 error: { _tag: "MyError", message: "BOOF" },
446 };
447 const exit = yield* Effect.exit(decodeRpcResult(envelope));
448 expect(Exit.isFailure(exit)).toBe(true);
449 if (Exit.isFailure(exit)) {
450 const failReason = exit.cause.reasons.find((r) => r._tag === "Fail");
451 expect(failReason).toBeDefined();
452 const error = (failReason as any).error as Record<string, unknown>;
453 expect(error._tag).toBe("MyError");
454 expect(error.message).toBe("BOOF");
455 }
456 }),
457 );
458
459 it.effect("fails with plain Error shape for error envelopes", () =>
460 Effect.gen(function* () {
461 const envelope: RpcErrorEnvelope = {
462 _tag: ErrorTag,
463 error: { name: "Error", message: "plain failure" },
464 };
465 const exit = yield* Effect.exit(decodeRpcResult(envelope));
466 expect(Exit.isFailure(exit)).toBe(true);
467 if (Exit.isFailure(exit)) {
468 const failReason = exit.cause.reasons.find((r) => r._tag === "Fail");
469 const error = (failReason as any).error as Record<string, unknown>;
470 expect(error.message).toBe("plain failure");
471 }
472 }),
473 );
474
475 it.effect("wraps stream envelopes in succeed (stream passthrough)", () =>
476 Effect.gen(function* () {
477 const body = textToReadableStream('{"s":1}\n');
478 const envelope: RpcStreamEnvelope = {
479 _tag: StreamTag,
480 encoding: "jsonl",
481 body,
482 };
483 const result = yield* decodeRpcResult(envelope);
484 expect(Stream.isStream(result)).toBe(true);
485 const chunks = yield* Stream.runCollect(result as Stream.Stream<any>);
486 expect(chunks).toEqual([{ s: 1 }]);
487 }),
488 );
489 },
490);
491
492// ---------------------------------------------------------------------------
493// toRpcStream
494// ---------------------------------------------------------------------------
495
496describe(
497 "toRpcStream",
498 {
499 tags: [
500 "unit",
501 "provider:cloudflare",
502 "provider:cloudflare:worker",
503 "local",
504 ],
505 },
506 () => {
507 it.effect("selects jsonl encoding for non-byte data", () =>
508 Effect.gen(function* () {
509 const stream = Stream.fromIterable([{ a: 1 }]);
510 const envelope = yield* toRpcStream(stream);
511 expect(envelope._tag).toBe(StreamTag);
512 expect(envelope.encoding).toBe("jsonl");
513 expect(envelope.body).toBeInstanceOf(ReadableStream);
514 }),
515 );
516
517 it.effect("roundtrips a single-element jsonl stream", () =>
518 Effect.gen(function* () {
519 const stream = Stream.fromIterable([{ a: 1 }]);
520 const envelope = yield* toRpcStream(stream);
521 const decoded = fromRpcReadableStream(envelope.body, "jsonl");
522 const chunks = yield* Stream.runCollect(decoded);
523 expect(chunks).toEqual([{ a: 1 }]);
524 }),
525 );
526
527 it.effect("selects bytes encoding for Uint8Array data", () =>
528 Effect.gen(function* () {
529 const data = new Uint8Array([1, 2, 3]);
530 const stream = Stream.fromIterable([data]);
531 const envelope = yield* toRpcStream(stream);
532 expect(envelope._tag).toBe(StreamTag);
533 expect(envelope.encoding).toBe("bytes");
534 expect(envelope.body).toBeInstanceOf(ReadableStream);
535 }),
536 );
537
538 it.effect("roundtrips a single-element bytes stream", () =>
539 Effect.gen(function* () {
540 const data = new Uint8Array([1, 2, 3]);
541 const stream = Stream.fromIterable([data]);
542 const envelope = yield* toRpcStream(stream);
543 const decoded = fromRpcReadableStream(envelope.body, "bytes");
544 const chunks = yield* Stream.runCollect(decoded);
545 expect(chunks).toEqual([data]);
546 }),
547 );
548
549 it.effect("handles empty stream as jsonl", () =>
550 Effect.gen(function* () {
551 const stream = Stream.empty;
552 const envelope = yield* toRpcStream(stream);
553 expect(envelope._tag).toBe(StreamTag);
554 expect(envelope.encoding).toBe("jsonl");
555
556 const decoded = fromRpcReadableStream(envelope.body, envelope.encoding);
557 const chunks = yield* Stream.runCollect(decoded);
558 expect(chunks).toEqual([]);
559 }),
560 );
561 },
562);
563
564// ---------------------------------------------------------------------------
565// makeRpcStub
566// ---------------------------------------------------------------------------
567
568describe(
569 "makeRpcStub",
570 {
571 tags: [
572 "unit",
573 "provider:cloudflare",
574 "provider:cloudflare:worker",
575 "local",
576 ],
577 },
578 () => {
579 it.effect("proxies successful calls", () =>
580 Effect.gen(function* () {
581 const mockStub = {
582 greet: async (name: string) => `hello ${name}`,
583 };
584 const stub = makeRpcStub<{
585 greet: (name: string) => Effect.Effect<string>;
586 }>(mockStub);
587
588 const result = yield* stub.greet("world");
589 expect(result).toBe("hello world");
590 }),
591 );
592
593 it.effect("wraps rejected promises as RpcCallError", () =>
594 Effect.gen(function* () {
595 const mockStub = {
596 boom: async () => {
597 throw new Error("kaboom");
598 },
599 };
600 const stub = makeRpcStub<{
601 boom: () => Effect.Effect<never, RpcCallError>;
602 }>(mockStub);
603
604 const exit = yield* Effect.exit(stub.boom());
605 expect(Exit.isFailure(exit)).toBe(true);
606 if (Exit.isFailure(exit)) {
607 const failReason = exit.cause.reasons.find((r) => r._tag === "Fail");
608 expect(failReason).toBeDefined();
609 const error = (failReason as any).error;
610 expect(error._tag).toBe("RpcCallError");
611 expect(error.method).toBe("boom");
612 }
613 }),
614 );
615
616 it.effect("decodes error envelopes into Effect.fail", () =>
617 Effect.gen(function* () {
618 const mockStub = {
619 failMe: async () => ({
620 _tag: ErrorTag,
621 error: { _tag: "MyError", message: "remote fail" },
622 }),
623 };
624 const stub = makeRpcStub<{
625 failMe: () => Effect.Effect<never, { _tag: string; message: string }>;
626 }>(mockStub);
627
628 const exit = yield* Effect.exit(stub.failMe());
629 expect(Exit.isFailure(exit)).toBe(true);
630 if (Exit.isFailure(exit)) {
631 const failReason = exit.cause.reasons.find((r) => r._tag === "Fail");
632 const error = (failReason as any).error as Record<string, unknown>;
633 expect(error._tag).toBe("MyError");
634 expect(error.message).toBe("remote fail");
635 }
636 }),
637 );
638
639 it.effect("decodes stream envelopes from successful calls", () =>
640 Effect.gen(function* () {
641 const body = textToReadableStream('{"n":42}\n');
642 const mockStub = {
643 streamMe: async () => ({
644 _tag: StreamTag,
645 encoding: "jsonl" as const,
646 body,
647 }),
648 };
649 const stub = makeRpcStub<{
650 streamMe: () => Effect.Effect<Stream.Stream<{ n: number }>>;
651 }>(mockStub);
652
653 const stream = yield* stub.streamMe();
654 expect(Stream.isStream(stream)).toBe(true);
655 const chunks = yield* Stream.runCollect(stream);
656 expect(chunks).toEqual([{ n: 42 }]);
657 }),
658 );
659 },
660);
661
662// ---------------------------------------------------------------------------
663// Stream error transport
664// ---------------------------------------------------------------------------
665
666describe(
667 "stream errors",
668 {
669 tags: [
670 "unit",
671 "provider:cloudflare",
672 "provider:cloudflare:worker",
673 "local",
674 ],
675 },
676 () => {
677 it.effect("toRpcStream encodes a stream that fails immediately", () =>
678 Effect.gen(function* () {
679 const stream = Stream.fail(new MyError({ message: "immediate" }));
680 const envelope = yield* toRpcStream(stream);
681 expect(envelope._tag).toBe(StreamTag);
682 expect(envelope.encoding).toBe("jsonl");
683
684 const decoded = fromRpcReadableStream(envelope.body, "jsonl");
685 const exit = yield* Effect.exit(Stream.runCollect(decoded));
686 expect(Exit.isFailure(exit)).toBe(true);
687 if (Exit.isFailure(exit)) {
688 const reason = exit.cause.reasons.find(Cause.isFailReason);
689 expect(reason).toBeDefined();
690 const err = reason!.error;
691 expect(err).toBeInstanceOf(RpcRemoteStreamError);
692 expect((err as RpcRemoteStreamError).error).toEqual({
693 _tag: "MyError",
694 message: "immediate",
695 });
696 }
697 }),
698 );
699
700 it.effect("toRpcStream encodes a stream that fails after elements", () =>
701 Effect.gen(function* () {
702 const stream = Stream.make(1, 2).pipe(
703 Stream.concat(Stream.fail(new MyError({ message: "mid" }))),
704 );
705 const envelope = yield* toRpcStream(stream);
706 expect(envelope._tag).toBe(StreamTag);
707 expect(envelope.encoding).toBe("jsonl");
708
709 const decoded = fromRpcReadableStream(envelope.body, "jsonl");
710 const exit = yield* Effect.exit(Stream.runCollect(decoded));
711 expect(Exit.isFailure(exit)).toBe(true);
712 if (Exit.isFailure(exit)) {
713 const reason = exit.cause.reasons.find(Cause.isFailReason);
714 expect(reason).toBeDefined();
715 const err = reason!.error;
716 expect(err).toBeInstanceOf(RpcRemoteStreamError);
717 expect((err as RpcRemoteStreamError).error).toEqual({
718 _tag: "MyError",
719 message: "mid",
720 });
721 }
722 }),
723 );
724
725 it.effect("fromRpcReadableStream decodes error marker in JSONL", () =>
726 Effect.gen(function* () {
727 const errorLine = JSON.stringify({
728 _tag: StreamErrorTag,
729 error: { _tag: "MyError", message: "wire" },
730 });
731 const body = textToReadableStream(`${errorLine}\n`);
732 const stream = fromRpcReadableStream(body, "jsonl");
733 const exit = yield* Effect.exit(Stream.runCollect(stream));
734 expect(Exit.isFailure(exit)).toBe(true);
735 if (Exit.isFailure(exit)) {
736 const reason = exit.cause.reasons.find(Cause.isFailReason);
737 const err = reason!.error;
738 expect(err).toBeInstanceOf(RpcRemoteStreamError);
739 expect((err as RpcRemoteStreamError).error).toEqual({
740 _tag: "MyError",
741 message: "wire",
742 });
743 }
744 }),
745 );
746
747 it.effect("fromRpcReadableStream yields elements before error marker", () =>
748 Effect.gen(function* () {
749 const errorLine = JSON.stringify({
750 _tag: StreamErrorTag,
751 error: { message: "after elements" },
752 });
753 const body = textToReadableStream(`{"v":1}\n{"v":2}\n${errorLine}\n`);
754 const stream = fromRpcReadableStream(body, "jsonl");
755 const collected: unknown[] = [];
756 const exit = yield* Effect.exit(
757 Stream.runForEach(stream, (item) =>
758 Effect.sync(() => {
759 collected.push(item);
760 }),
761 ),
762 );
763 expect(collected).toEqual([{ v: 1 }, { v: 2 }]);
764 expect(Exit.isFailure(exit)).toBe(true);
765 }),
766 );
767
768 it.effect(
769 "makeRpcStub preserves stream errors (not collapsed to RpcCallError)",
770 () =>
771 Effect.gen(function* () {
772 const errorLine = JSON.stringify({
773 _tag: StreamErrorTag,
774 error: { _tag: "MyError", message: "remote stream err" },
775 });
776 const body = textToReadableStream(`{"n":1}\n${errorLine}\n`);
777 const mockStub = {
778 streamFail: async () => ({
779 _tag: StreamTag,
780 encoding: "jsonl" as const,
781 body,
782 }),
783 };
784 const stub = makeRpcStub<{
785 streamFail: () => Effect.Effect<
786 Stream.Stream<unknown, RpcRemoteStreamError>
787 >;
788 }>(mockStub);
789
790 const stream = yield* stub.streamFail();
791 const exit = yield* Effect.exit(Stream.runCollect(stream));
792 expect(Exit.isFailure(exit)).toBe(true);
793 if (Exit.isFailure(exit)) {
794 const reason = exit.cause.reasons.find(Cause.isFailReason);
795 expect(reason).toBeDefined();
796 expect(reason!.error).toBeInstanceOf(RpcRemoteStreamError);
797 }
798 }),
799 );
800 },
801);