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 // A stopped worker never drains; its exit settles the race instead of a hang.
45 yield* Effect.raceFirst(worker.drain, Fiber.await(yield* Deferred.await(workerFiber)));
46
47 expect(processed).toEqual(["ok"]);
48 // The item that interrupted itself was cancelled, not failed.
49 expect(loggedErrors).toEqual(["typed failure", "defect"]);
50 }),
51 ).pipe(Effect.provide(loggerLayer));
52 });
53
54 it.effect("stops quietly when its scope closes during an item", () => {
55 const { loggedErrors, loggerLayer } = captureErrorLogs();
56
57 return Effect.gen(function* () {
58 const scope = yield* Scope.make();
59 const processed: string[] = [];
60 const workerFiber = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>();
61 const worker = yield* makeDrainableWorker((item: string) =>
62 Effect.gen(function* () {
63 processed.push(item);
64 yield* Deferred.succeed(workerFiber, yield* Effect.fiber);
65 return yield* Effect.never;
66 }),
67 ).pipe(Scope.provide(scope));
68
69 yield* worker.enqueue("block");
70 yield* worker.enqueue("later");
71 const fiber = yield* Deferred.await(workerFiber);
72 yield* Scope.close(scope, Exit.void);
73
74 expect(Exit.hasInterrupts(yield* Fiber.await(fiber))).toBe(true);
75 expect(processed).toEqual(["block"]);
76 expect(loggedErrors).toEqual([]);
77
78 // Queued and late items are dropped with the queue, so drain resolves at once.
79 yield* worker.enqueue("after close");
80 const drained = yield* Effect.forkChild(worker.drain, { startImmediately: true });
81 expect(drained.pollUnsafe()).toEqual(Exit.void);
82 expect(processed).toEqual(["block"]);
83 }).pipe(Effect.provide(loggerLayer));
84 });
85
86 it.live("waits for work enqueued during active processing before draining", () =>
87 Effect.scoped(
88 Effect.gen(function* () {
89 const processed: string[] = [];
90 const firstStarted = yield* Deferred.make<void>();
91 const releaseFirst = yield* Deferred.make<void>();
92 const secondStarted = yield* Deferred.make<void>();
93 const releaseSecond = yield* Deferred.make<void>();
94
95 const worker = yield* makeDrainableWorker((item: string) =>
96 Effect.gen(function* () {
97 if (item === "first") {
98 yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
99 yield* Deferred.await(releaseFirst);
100 }
101
102 if (item === "second") {
103 yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie);
104 yield* Deferred.await(releaseSecond);
105 }
106
107 processed.push(item);
108 }),
109 );
110
111 yield* worker.enqueue("first");
112 yield* Deferred.await(firstStarted);
113
114 const drained = yield* Deferred.make<void>();
115 yield* Effect.forkChild(
116 worker.drain.pipe(
117 Effect.tap(() => Deferred.succeed(drained, undefined).pipe(Effect.orDie)),
118 ),
119 );
120
121 yield* worker.enqueue("second");
122 yield* Deferred.succeed(releaseFirst, undefined);
123 yield* Deferred.await(secondStarted);
124
125 expect(yield* Deferred.isDone(drained)).toBe(false);
126
127 yield* Deferred.succeed(releaseSecond, undefined);
128 yield* Deferred.await(drained);
129
130 expect(processed).toEqual(["first", "second"]);
131 }),
132 ),
133 );
134});