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