Open History to see the last commit
1import { it } from "@effect/vitest";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as PubSub from "effect/PubSub";
6import * as Queue from "effect/Queue";
7import * as Stream from "effect/Stream";
8import { describe, expect } from "vite-plus/test";
9
10import {
11 bufferLatestLiveStream,
12 makeLiveStreamBudget,
13 replayAndBufferProjectedLiveEvents,
14 type RetainedLiveItem,
15} from "./LiveStreamBudget.ts";
16
17describe("LiveStreamBudget", () => {
18 it.effect("closes the source without releasing a batch still waiting for an ACK", () =>
19 Effect.scoped(
20 Effect.gen(function* () {
21 const budget = yield* makeLiveStreamBudget({ maxItems: 3 });
22 const queue = yield* Queue.unbounded<RetainedLiveItem<{ text: string }>>();
23 const sourceClosed = yield* Deferred.make<void>();
24 const first = yield* budget.retain({ text: "first" });
25 const second = yield* budget.retain({ text: "second" });
26 const third = yield* budget.retain({ text: "third" });
27 yield* Queue.offerAll(queue, [first, second, third]);
28 yield* Effect.scoped(
29 Effect.gen(function* () {
30 const pull = yield* Stream.toPull(
31 budget.deliver(
32 Stream.fromQueue(queue).pipe(
33 Stream.rechunk(1),
34 Stream.ensuring(Deferred.succeed(sourceClosed, undefined)),
35 ),
36 ),
37 );
38 expect(yield* pull).toEqual([{ text: "first" }]);
39 expect(yield* Queue.size(queue)).toBe(0);
40 expect((yield* budget.usage).retainedItems).toBe(3);
41 const overflow = yield* budget.retain({ text: "fourth" }).pipe(Effect.result);
42 expect(overflow._tag).toBe("Failure");
43 yield* Deferred.await(sourceClosed);
44 yield* budget.closed;
45 expect((yield* budget.usage).retainedItems).toBe(1);
46 }),
47 );
48 expect(yield* budget.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
49 }),
50 ),
51 );
52});
53
54type Event = { readonly sequence: number; readonly text: string; readonly threadId?: string };
55
56describe("bufferLatestLiveStream", () => {
57 it.effect.each(["items", "bytes"] as const)(
58 "counts an unacknowledged state against the %s limit when its replacement arrives",
59 (limit) =>
60 Effect.scoped(
61 Effect.gen(function* () {
62 const releaseUpdate = yield* Deferred.make<void>();
63 const closed = yield* Deferred.make<void>();
64 const first = { sequence: 1, text: "é".repeat(20) };
65 const next = { ...first, sequence: 2 };
66 const pull = yield* Stream.toPull(
67 bufferLatestLiveStream(
68 Stream.concat(
69 Stream.succeed(first),
70 Stream.fromEffect(Deferred.await(releaseUpdate)).pipe(Stream.map(() => next)),
71 ).pipe(Stream.ensuring(Deferred.succeed(closed, undefined))),
72 () => "same-thread",
73 limit === "items"
74 ? { maxItems: 1 }
75 : { maxSerializedBytes: Buffer.byteLength(JSON.stringify(first)) },
76 ),
77 );
78 expect(yield* pull).toEqual([first]);
79 yield* Deferred.succeed(releaseUpdate, undefined);
80 yield* Deferred.await(closed);
81 const result = yield* pull.pipe(Effect.result);
82 expect(result._tag).toBe("Failure");
83 if (result._tag === "Failure") expect(result.failure._tag).toBe("LiveStreamBufferError");
84 }),
85 ),
86 );
87});
88
89describe("replayAndBufferProjectedLiveEvents", () => {
90 it.effect.each(["items", "bytes"] as const)(
91 "replays the undelivered tail after a %s overflow without losing or duplicating events",
92 (limit) =>
93 Effect.scoped(
94 Effect.gen(function* () {
95 const pubsub = yield* PubSub.unbounded<Event>();
96 const subscribed = yield* Deferred.make<PubSub.Subscription<Event>>();
97 const events = [
98 { sequence: 1, text: "one" },
99 { sequence: 2, text: "two" },
100 { sequence: 3, text: "three" },
101 ];
102 const limits = limit === "items" ? { maxItems: 1 } : { maxSerializedBytes: 40 };
103 const pull = yield* Stream.toPull(
104 replayAndBufferProjectedLiveEvents(
105 {
106 project: (event: Event) => event,
107 subscribe: PubSub.subscribe(pubsub).pipe(
108 Effect.tap((subscription) => Deferred.succeed(subscribed, subscription)),
109 ),
110 latestSequence: Effect.succeed(1),
111 replay: () => Stream.succeed(events[0]!),
112 },
113 limits,
114 ),
115 );
116 const received = [...(yield* pull)];
117 // Hold the ACK while more durable events arrive.
118 yield* PubSub.publishAll(pubsub, events.slice(1));
119 yield* (yield* Deferred.await(subscribed)).shutdownHook.await;
120 const overflow = yield* pull.pipe(Effect.result);
121 expect(overflow._tag).toBe("Failure");
122 if (overflow._tag === "Failure")
123 expect(overflow.failure._tag).toBe("LiveStreamBufferError");
124
125 const afterSequence = received.at(-1)!.sequence;
126 const resumed = yield* replayAndBufferProjectedLiveEvents(
127 {
128 project: (event: Event) => event,
129 subscribe: PubSub.subscribe(pubsub),
130 afterSequence,
131 latestSequence: Effect.succeed(3),
132 replay: (throughSequence) =>
133 Stream.fromIterable(
134 events.filter(
135 (event) => event.sequence > afterSequence && event.sequence <= throughSequence,
136 ),
137 ),
138 },
139 limits,
140 ).pipe(Stream.take(2), Stream.runCollect);
141 expect([...received, ...resumed]).toEqual(events);
142 expect(yield* PubSub.size(pubsub)).toBe(0);
143 }),
144 ),
145 );
146
147 it.effect.each(["high-water", "replay"] as const)(
148 "unsubscribes and cancels a blocked %s read on live overflow",
149 (phase) =>
150 Effect.scoped(
151 Effect.gen(function* () {
152 const pubsub = yield* PubSub.unbounded<Event>();
153 const subscribed = yield* Deferred.make<PubSub.Subscription<Event>>();
154 const readStarted = yield* Deferred.make<void>();
155 const readClosed = yield* Deferred.make<void>();
156 const blockedRead = Deferred.succeed(readStarted, undefined).pipe(
157 Effect.andThen(Effect.never),
158 Effect.ensuring(Deferred.succeed(readClosed, undefined)),
159 );
160 const pull = yield* Stream.toPull(
161 replayAndBufferProjectedLiveEvents(
162 {
163 project: (event: Event) => event,
164 subscribe: PubSub.subscribe(pubsub).pipe(
165 Effect.tap((subscription) => Deferred.succeed(subscribed, subscription)),
166 ),
167 latestSequence:
168 phase === "high-water" ? blockedRead.pipe(Effect.as(0)) : Effect.succeed(0),
169 replay: () => (phase === "replay" ? Stream.fromEffect(blockedRead) : Stream.empty),
170 },
171 { maxItems: 2 },
172 ),
173 );
174 const reader = yield* pull.pipe(Effect.result, Effect.forkScoped);
175 yield* Deferred.await(readStarted);
176 yield* PubSub.publishAll(pubsub, [
177 { sequence: 1, text: "one" },
178 { sequence: 2, text: "two" },
179 { sequence: 3, text: "overflow" },
180 ]);
181 yield* (yield* Deferred.await(subscribed)).shutdownHook.await;
182 yield* Deferred.await(readClosed);
183 const result = yield* Fiber.join(reader);
184 expect(result._tag).toBe("Failure");
185 if (result._tag === "Failure") expect(result.failure._tag).toBe("LiveStreamBufferError");
186 expect(yield* PubSub.size(pubsub)).toBe(0);
187 yield* PubSub.publish(pubsub, { sequence: 4, text: "after overflow" });
188 expect(yield* PubSub.size(pubsub)).toBe(0);
189 }),
190 ),
191 );
192
193 it.effect("lets a fast reader replay a database page larger than both buffer limits", () =>
194 Effect.scoped(
195 Effect.gen(function* () {
196 const pubsub = yield* PubSub.unbounded<Event>();
197 const events = Array.from({ length: 20 }, (_, index) => ({
198 sequence: index + 1,
199 text: "historical event",
200 }));
201 const replayed = yield* replayAndBufferProjectedLiveEvents(
202 {
203 project: (event: Event) => event,
204 subscribe: PubSub.subscribe(pubsub),
205 latestSequence: Effect.succeed(20),
206 replay: () => Stream.fromIterable(events),
207 },
208 { maxItems: 2, maxSerializedBytes: 100 },
209 ).pipe(Stream.take(20), Stream.runCollect);
210 expect(replayed).toEqual(events);
211 expect(yield* PubSub.size(pubsub)).toBe(0);
212 }),
213 ),
214 );
215
216 it.effect(
217 "filters unrelated events before buffering and deduplicates the replay/live overlap",
218 () =>
219 Effect.scoped(
220 Effect.gen(function* () {
221 const pubsub = yield* PubSub.unbounded<Event>();
222 const highWaterStarted = yield* Deferred.make<void>();
223 const highWater = yield* Deferred.make<number>();
224 const first = { sequence: 1, text: "first", threadId: "selected" };
225 const unrelatedReplay = { sequence: 0, text: "unrelated replay", threadId: "other" };
226 const overlap = { sequence: 2, text: "overlap", threadId: "selected" };
227 const last = { sequence: 10, text: "last", threadId: "selected" };
228 const reader = yield* replayAndBufferProjectedLiveEvents(
229 {
230 project: (event: Event) => event,
231 subscribe: PubSub.subscribe(pubsub),
232 latestSequence: Deferred.succeed(highWaterStarted, undefined).pipe(
233 Effect.andThen(Deferred.await(highWater)),
234 ),
235 replay: () => Stream.make(unrelatedReplay, first, overlap),
236 filter: (event) => event.threadId === "selected",
237 },
238 { maxItems: 4 },
239 ).pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped);
240 yield* Deferred.await(highWaterStarted);
241 yield* PubSub.publishAll(pubsub, [
242 overlap,
243 ...Array.from({ length: 6 }, (_, index) => ({
244 sequence: index + 3,
245 text: "unrelated",
246 threadId: "other",
247 })),
248 last,
249 ]);
250 yield* Deferred.succeed(highWater, 2);
251 expect(yield* Fiber.join(reader)).toEqual([first, overlap, last]);
252 expect(yield* PubSub.size(pubsub)).toBe(0);
253 }),
254 ),
255 );
256});