1import { expect, it } from "@effect/vitest";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as Queue from "effect/Queue";
6import * as Socket from "effect/socket/Socket";
7import { bridge } from "./OutboundSocket.ts";
8
9type Frame = string | Uint8Array;
10const endpoint = (incoming: Queue.Queue<Frame>, write: (frame: Frame) => Effect.Effect<void>) => {
11 const send = (frame: Frame | Socket.CloseEvent) =>
12 Socket.isCloseEvent(frame) ? Effect.void : write(frame);
13 return Socket.make({
14 reader: Effect.succeed({
15 pull: Queue.take(incoming).pipe(Effect.map((frame) => [frame] as const)),
16 upgrade: () => Effect.void,
17 }),
18 writer: Effect.succeed({
19 write: send,
20 writeAll: (frames) => Effect.forEach(frames, send, { discard: true }),
21 }),
22 });
23};
24const tunnelPair = Effect.gen(function* () {
25 const left = yield* Queue.unbounded<Frame>();
26 const right = yield* Queue.unbounded<Frame>();
27 const sent: Uint8Array[] = [];
28 const full = yield* Deferred.make<void>();
29 return {
30 left: endpoint(left, (frame) =>
31 Effect.gen(function* () {
32 if (typeof frame !== "string" && frame[0] === 0) {
33 sent.push(frame);
34 if (sent.length === 16) yield* Deferred.succeed(full, undefined);
35 }
36 yield* Queue.offer(right, frame);
37 }),
38 ),
39 right: endpoint(right, (frame) => Queue.offer(left, frame).pipe(Effect.asVoid)),
40 injectLeft: (frame: Frame) => Queue.offer(left, frame),
41 sent,
42 full,
43 };
44});
45const tcp = (expectedBytes: number, stalled?: Deferred.Deferred<void>) =>
46 Effect.gen(function* () {
47 const outgoing = yield* Queue.unbounded<Frame>();
48 const received: Uint8Array[] = [];
49 const done = yield* Deferred.make<void>();
50 const socket = endpoint(outgoing, (frame) =>
51 Effect.gen(function* () {
52 if (stalled) yield* Deferred.await(stalled);
53 received.push(typeof frame === "string" ? new TextEncoder().encode(frame) : frame);
54 if (received.reduce((size, chunk) => size + chunk.byteLength, 0) >= expectedBytes)
55 yield* Deferred.succeed(done, undefined);
56 }),
57 );
58 return { socket, received, done, send: (frame: Frame) => Queue.offer(outgoing, frame) };
59 });
60
61it.effect(
62 "bounds a stalled tunnel to sixteen 64KiB payloads and releases a sender waiting for credit on cancellation",
63 () =>
64 Effect.gen(function* () {
65 const pair = yield* tunnelPair;
66 const left = yield* tcp(0);
67 const gate = yield* Deferred.make<void>();
68 const right = yield* tcp(1, gate);
69 const finalized = yield* Deferred.make<void>();
70 const sender = yield* bridge(left.socket, pair.left).pipe(
71 Effect.ensuring(Deferred.succeed(finalized, undefined)),
72 Effect.forkScoped({ startImmediately: true }),
73 );
74 yield* bridge(right.socket, pair.right).pipe(Effect.forkScoped({ startImmediately: true }));
75 yield* left.send(new Uint8Array(64 * 1024 * 40));
76 yield* Deferred.await(pair.full);
77 expect(pair.sent).toHaveLength(16);
78 expect(pair.sent.reduce((size, frame) => size + frame.byteLength - 5, 0)).toBe(1024 * 1024);
79 expect(pair.sent.every((frame) => frame.byteLength <= 64 * 1024 + 5)).toBe(true);
80 yield* Fiber.interrupt(sender);
81 expect(yield* Deferred.isDone(finalized)).toBe(true);
82 }),
83);
84
85it.effect("streams bursts in both directions without credit deadlock or byte loss", () =>
86 Effect.gen(function* () {
87 const pair = yield* tunnelPair;
88 const bytes = new Uint8Array(64 * 1024 * 40).map((_, index) => index % 251);
89 const reverse = bytes.toReversed();
90 const left = yield* tcp(bytes.length);
91 const right = yield* tcp(bytes.length);
92 yield* bridge(left.socket, pair.left).pipe(Effect.forkScoped({ startImmediately: true }));
93 yield* bridge(right.socket, pair.right).pipe(Effect.forkScoped({ startImmediately: true }));
94 yield* left.send(bytes);
95 yield* right.send(reverse);
96 yield* Deferred.await(left.done);
97 yield* Deferred.await(right.done);
98 expect(Buffer.concat(left.received)).toEqual(Buffer.from(reverse));
99 expect(Buffer.concat(right.received)).toEqual(Buffer.from(bytes));
100 }),
101);
102
103it.effect("rejects unsolicited acknowledgments instead of manufacturing send credit", () =>
104 Effect.gen(function* () {
105 const pair = yield* tunnelPair;
106 const local = yield* tcp(0);
107 const running = yield* bridge(local.socket, pair.left).pipe(
108 Effect.result,
109 Effect.forkScoped({ startImmediately: true }),
110 );
111 yield* pair.injectLeft(new Uint8Array([1, 0, 0, 0, 0]));
112 expect((yield* Fiber.join(running))._tag).toBe("Failure");
113 }),
114);