1import { it } from "@effect/vitest";
2import { describe, expect } from "vite-plus/test";
3import * as Cause from "effect/Cause";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
8import * as Logger from "effect/Logger";
9import * as Scope from "effect/Scope";
10
11import { makeDrainableWorker } from "./DrainableWorker.ts";
12
13const captureErrorLogs = () => {
14 const loggedErrors: Array<unknown> = [];
15 const logger = Logger.make(({ logLevel, cause }) => {
16 if (logLevel === "Error") loggedErrors.push(Cause.squash(cause));
17 });
18 return { loggedErrors, loggerLayer: Logger.layer([logger], { mergeWithExisting: false }) };
19};
20
21describe("makeDrainableWorker", () => {
22 it.effect("logs a failed or defective item and keeps processing later items", () => {
23 const { loggedErrors, loggerLayer } = captureErrorLogs();
24
25 return Effect.scoped(
26 Effect.gen(function* () {
27 const processed: string[] = [];
28 const workerFiber = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>();
29 const worker = yield* makeDrainableWorker((item: string) =>
30 Effect.gen(function* () {
31 yield* Deferred.succeed(workerFiber, yield* Effect.fiber);
32 if (item === "fail") return yield* Effect.fail("typed failure");
33 if (item === "die") return yield* Effect.die("defect");
34 if (item === "interrupt") return yield* Effect.interrupt;
35 processed.push(item);
36 }),
37 );
38
39 yield* worker.enqueue("fail");
40 yield* worker.enqueue("die");
41 yield* worker.enqueue("interrupt");
42 yield* worker.enqueue("ok");
43
44 yield* Effect.raceFirst(worker.drain, Fiber.await(yield* Deferred.await(workerFiber)));
45
46 expect(processed).toEqual(["ok"]);
47 expect(loggedErrors).toEqual(["typed failure", "defect"]);
48 }),
49 ).pipe(Effect.provide(loggerLayer));
50 });
51
52 it.effect("stops quietly when its scope closes during an item", () => {
53 const { loggedErrors, loggerLayer } = captureErrorLogs();
54
55 return Effect.gen(function* () {
56 const scope = yield* Scope.make();
57 const processed: string[] = [];
58 const workerFiber = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>();
59 const worker = yield* makeDrainableWorker((item: string) =>
60 Effect.gen(function* () {
61 processed.push(item);
62 yield* Deferred.succeed(workerFiber, yield* Effect.fiber);
63 return yield* Effect.never;
64 }),
65 ).pipe(Scope.provide(scope));
66
67 yield* worker.enqueue("block");
68 yield* worker.enqueue("later");
69 const fiber = yield* Deferred.await(workerFiber);
70 yield* Scope.close(scope, Exit.void);
71
72 expect(Exit.hasInterrupts(yield* Fiber.await(fiber))).toBe(true);
73 expect(processed).toEqual(["block"]);
74 expect(loggedErrors).toEqual([]);
75
76 yield* worker.enqueue("after close");
77 const drained = yield* Effect.forkChild(worker.drain, { startImmediately: true });
78 expect(drained.pollUnsafe()).toEqual(Exit.void);
79 expect(processed).toEqual(["block"]);
80 }).pipe(Effect.provide(loggerLayer));
81 });
82
83 it.effect("waits for work enqueued during active processing before draining", () =>
84 Effect.gen(function* () {
85 const processed: string[] = [];
86 const firstStarted = yield* Deferred.make<void>();
87 const releaseFirst = yield* Deferred.make<void>();
88 const secondStarted = yield* Deferred.make<void>();
89 const releaseSecond = yield* Deferred.make<void>();
90
91 const worker = yield* makeDrainableWorker((item: string) =>
92 Effect.gen(function* () {
93 if (item === "first") {
94 yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
95 yield* Deferred.await(releaseFirst);
96 }
97
98 if (item === "second") {
99 yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie);
100 yield* Deferred.await(releaseSecond);
101 }
102
103 processed.push(item);
104 }),
105 );
106
107 yield* worker.enqueue("first");
108 yield* Deferred.await(firstStarted);
109
110 const drained = yield* Deferred.make<void>();
111 yield* Effect.forkChild(
112 worker.drain.pipe(
113 Effect.tap(() => Deferred.succeed(drained, undefined).pipe(Effect.orDie)),
114 ),
115 );
116
117 yield* worker.enqueue("second");
118 yield* Deferred.succeed(releaseFirst, undefined);
119 yield* Deferred.await(secondStarted);
120
121 expect(yield* Deferred.isDone(drained)).toBe(false);
122
123 yield* Deferred.succeed(releaseSecond, undefined);
124 yield* Deferred.await(drained);
125
126 expect(processed).toEqual(["first", "second"]);
127 }),
128 );
129});