1import { it } from "@effect/vitest";
2import { describe, expect } from "vite-plus/test";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5
6import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts";
7
8describe("makeKeyedCoalescingWorker", () => {
9 it.effect("lets a queued key drain before a continuously updated key finishes", () =>
10 Effect.gen(function* () {
11 const processed: string[] = [];
12 const busyStarted = yield* Deferred.make<void>();
13 const releaseBusy = yield* Deferred.make<void>();
14
15 const worker = yield* makeKeyedCoalescingWorker<string, number, never, never>({
16 merge: (_current, next) => next,
17 process: (key, value): Effect.Effect<void> =>
18 Effect.gen(function* () {
19 processed.push(`${key}:${value}`);
20 if (key !== "busy") return;
21
22 if (value === 1) {
23 yield* Deferred.succeed(busyStarted, undefined);
24 yield* Deferred.await(releaseBusy);
25 }
26 if (value < 4) yield* worker.enqueue(key, value + 1);
27 }),
28 });
29
30 yield* worker.enqueue("busy", 1);
31 yield* Deferred.await(busyStarted);
32 yield* worker.enqueue("quiet", 1);
33 yield* Deferred.succeed(releaseBusy, undefined);
34 yield* worker.drainKey("quiet");
35 yield* worker.drainKey("busy");
36
37 expect(processed).toEqual(["busy:1", "quiet:1", "busy:2", "busy:3", "busy:4"]);
38 }),
39 );
40
41 it.effect("merges updates while waiting behind another key and drains the merged batch", () =>
42 Effect.gen(function* () {
43 const processed: string[] = [];
44 const busyStarted = yield* Deferred.make<void>();
45 const releaseBusy = yield* Deferred.make<void>();
46 const quietStarted = yield* Deferred.make<void>();
47 const releaseQuiet = yield* Deferred.make<void>();
48 const mergedStarted = yield* Deferred.make<void>();
49 const releaseMerged = yield* Deferred.make<void>();
50
51 const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
52 merge: (current, next) => `${current}+${next}`,
53 process: (key, value) =>
54 Effect.gen(function* () {
55 processed.push(`${key}:${value}`);
56 if (value === "first") {
57 yield* Deferred.succeed(busyStarted, undefined);
58 yield* Deferred.await(releaseBusy);
59 } else if (key === "quiet") {
60 yield* Deferred.succeed(quietStarted, undefined);
61 yield* Deferred.await(releaseQuiet);
62 } else if (value === "a+b+c") {
63 yield* Deferred.succeed(mergedStarted, undefined);
64 yield* Deferred.await(releaseMerged);
65 }
66 }),
67 });
68
69 yield* worker.enqueue("busy", "first");
70 yield* Deferred.await(busyStarted);
71 yield* worker.enqueue("quiet", "quiet");
72 yield* worker.enqueue("busy", "a");
73 yield* Deferred.succeed(releaseBusy, undefined);
74 yield* Deferred.await(quietStarted);
75 expect(processed).toEqual(["busy:first", "quiet:quiet"]);
76 yield* worker.enqueue("busy", "b");
77 yield* worker.enqueue("busy", "c");
78
79 const drained = yield* Deferred.make<void>();
80 yield* Effect.forkChild(
81 worker.drainKey("busy").pipe(Effect.andThen(Deferred.succeed(drained, undefined))),
82 { startImmediately: true },
83 );
84 expect(yield* Deferred.isDone(drained)).toBe(false);
85
86 yield* Deferred.succeed(releaseQuiet, undefined);
87 yield* Deferred.await(mergedStarted);
88 expect(yield* Deferred.isDone(drained)).toBe(false);
89 yield* Deferred.succeed(releaseMerged, undefined);
90 yield* Deferred.await(drained);
91 yield* worker.drainKey("quiet");
92
93 expect(processed).toEqual(["busy:first", "quiet:quiet", "busy:a+b+c"]);
94 }),
95 );
96
97 it.effect("waits for latest work enqueued during active processing before draining the key", () =>
98 Effect.gen(function* () {
99 const processed: string[] = [];
100 const firstStarted = yield* Deferred.make<void>();
101 const releaseFirst = yield* Deferred.make<void>();
102 const secondStarted = yield* Deferred.make<void>();
103 const releaseSecond = yield* Deferred.make<void>();
104
105 const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
106 merge: (_current, next) => next,
107 process: (key, value) =>
108 Effect.gen(function* () {
109 processed.push(`${key}:${value}`);
110
111 if (value === "first") {
112 yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
113 yield* Deferred.await(releaseFirst);
114 }
115
116 if (value === "second") {
117 yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie);
118 yield* Deferred.await(releaseSecond);
119 }
120 }),
121 });
122
123 yield* worker.enqueue("terminal-1", "first");
124 yield* Deferred.await(firstStarted);
125
126 const drained = yield* Deferred.make<void>();
127 yield* Effect.forkChild(
128 worker
129 .drainKey("terminal-1")
130 .pipe(Effect.tap(() => Deferred.succeed(drained, undefined).pipe(Effect.orDie))),
131 );
132
133 yield* worker.enqueue("terminal-1", "second");
134 yield* Deferred.succeed(releaseFirst, undefined);
135 yield* Deferred.await(secondStarted);
136
137 expect(yield* Deferred.isDone(drained)).toBe(false);
138
139 yield* Deferred.succeed(releaseSecond, undefined);
140 yield* Deferred.await(drained);
141
142 expect(processed).toEqual(["terminal-1:first", "terminal-1:second"]);
143 }),
144 );
145
146 it.effect("requeues pending work for a key after a processor failure and keeps draining", () =>
147 Effect.gen(function* () {
148 const processed: string[] = [];
149 const firstStarted = yield* Deferred.make<void>();
150 const releaseFailure = yield* Deferred.make<void>();
151 const secondProcessed = yield* Deferred.make<void>();
152
153 const worker = yield* makeKeyedCoalescingWorker<string, string, string, never>({
154 merge: (_current, next) => next,
155 process: (key, value) =>
156 Effect.gen(function* () {
157 processed.push(`${key}:${value}`);
158
159 if (value === "first") {
160 yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
161 yield* Deferred.await(releaseFailure);
162 return yield* Effect.fail("boom");
163 }
164
165 if (value === "second") {
166 yield* Deferred.succeed(secondProcessed, undefined).pipe(Effect.orDie);
167 }
168 }),
169 });
170
171 yield* worker.enqueue("terminal-1", "first");
172 yield* Deferred.await(firstStarted);
173 yield* worker.enqueue("terminal-2", "quiet");
174 yield* worker.enqueue("terminal-1", "second");
175 yield* Deferred.succeed(releaseFailure, undefined);
176 yield* Deferred.await(secondProcessed);
177 yield* worker.drainKey("terminal-1");
178
179 expect(processed).toEqual(["terminal-1:first", "terminal-2:quiet", "terminal-1:second"]);
180 }),
181 );
182});