History

fix(server): release consumed event replay pages (#10777)

Guillermo Casanova committed 2 months ago

Commit 08463e21 parent 18626862 files · +82 −45Browse files

Unified · Split

2 files
  1. apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts +51 −0
  2. apps/server/src/persistence/Layers/OrchestrationEventStore.ts +31 −45
apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts+51−0
@@ -1,3 +1,5 @@
1+import * as NodeV8 from "node:v8";
2+
13 import {
24 CommandId,
35 EventId,
@@ -303,6 +305,55 @@ layer("OrchestrationEventStore", (it) => { · 299 unmodified lines
303305 replayed.map((event) => event.sequence),
304306 persisted.map((event) => event.sequence),
305307 );
308+ const limited = store.readFromSequence(persisted[0]!.sequence, 501.9);
309+ for (let run = 0; run < 2; run++) {
310+ assert.deepEqual(
311+ (yield* Stream.runCollect(limited)).map((event) => event.sequence),
312+ persisted.slice(1).map((event) => event.sequence),
313+ );
314+ }
315+ assert.deepEqual(yield* Stream.runCollect(store.readFromSequence(0, -1)), []);
306316 }),
307317 );
308318 });
319+
320+for (const reader of ["all", "aggregate"] as const) {
321+ it.effect(`releases consumed pages during ${reader} replay`, () =>
322+ Effect.gen(function* () {
323+ const store = yield* OrchestrationEventStore;
324+ const threadId = ThreadId.make(`retention-${reader}`);
325+ yield* Effect.forEach(
326+ Array.from({ length: 1_501 }, (_, index) => index),
327+ (index) => store.append(messageEvent(threadId, `retention-${reader}-${index}`)),
328+ { discard: true },
329+ );
330+ // oxlint-disable-next-line typescript/no-extraneous-class -- Identifies page markers for V8's heap query.
331+ class ReplayPage {}
332+ let count = 0;
333+ const replay =
334+ reader === "all"
335+ ? store.readAll()
336+ : store.readAggregateRange({
337+ aggregateKind: "thread",
338+ aggregateId: threadId,
339+ fromSequenceExclusive: 0,
340+ toSequenceInclusive: 1_501,
341+ limit: 1_501,
342+ });
343+ yield* Stream.runForEach(replay, (event) =>
344+ Effect.sync(() => {
345+ assert.equal(event.sequence, count + 1);
346+ if (count % 500 === 0) {
347+ // Count live page markers after full GC, without timing or heap-size thresholds.
348+ Object.assign(event, { replayPage: new ReplayPage() });
349+ assert.isAtMost(NodeV8.queryObjects(ReplayPage, { format: "count" }), 1);
350+ }
351+ count++;
352+ }),
353+ );
354+ assert.equal(count, 1_501);
355+ }).pipe(
356+ Effect.provide(OrchestrationEventStoreLive.pipe(Layer.provide(SqlitePersistenceMemory))),
357+ ),
358+ );
359+}
apps/server/src/persistence/Layers/OrchestrationEventStore.ts+31−45
@@ -287,11 +287,9 @@ const makeEventStore = Effect.gen(function* () { · 286 unmodified lines
287287 if (normalizedLimit === 0) {
288288 return Stream.empty;
289289 }
290− const readPage = (
291− cursor: number,
292− remaining: number,
293− ): Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError> =>
294− Stream.fromEffect(
290+ return Stream.paginate(
291+ { cursor: sequenceExclusive, remaining: normalizedLimit },
292+ ({ cursor, remaining }) =>
295293 readEventRowsFromSequence({
296294 sequenceExclusive: cursor,
297295 limit: Math.min(remaining, READ_PAGE_SIZE),
@@ -311,24 +309,18 @@ const makeEventStore = Effect.gen(function* () { · 13 unmodified lines
311309 ),
312310 ),
313311 ),
312+ Effect.map((events) => {
313+ const last = events.at(-1);
314+ const nextRemaining = remaining - events.length;
315+ return [
316+ events,
317+ last === undefined || nextRemaining <= 0
318+ ? Option.none()
319+ : Option.some({ cursor: last.sequence, remaining: nextRemaining }),
320+ ] as const;
321+ }),
314322 ),
315− ).pipe(
316− Stream.flatMap((events) => {
317− if (events.length === 0) {
318− return Stream.empty;
319− }
320− const nextRemaining = remaining - events.length;
321− if (nextRemaining <= 0) {
322− return Stream.fromIterable(events);
323− }
324− return Stream.concat(
325− Stream.fromIterable(events),
326− readPage(events[events.length - 1]!.sequence, nextRemaining),
327− );
328− }),
329− );
330−
331− return readPage(sequenceExclusive, normalizedLimit);
323+ );
332324 };
333325
334326 const findEventAfter = SqlSchema.findOneOption({
@@ -363,11 +355,9 @@ const makeEventStore = Effect.gen(function* () { · 28 unmodified lines
363355 if (limit === 0 || input.fromSequenceExclusive >= input.toSequenceInclusive) {
364356 return Stream.empty;
365357 }
366− const readPage = (
367− cursor: number,
368− remaining: number,
369− ): Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError> =>
370− Stream.fromEffect(
358+ return Stream.paginate(
359+ { cursor: input.fromSequenceExclusive, remaining: limit },
360+ ({ cursor, remaining }) =>
371361 readAggregateEventRows({
372362 ...input,
373363 fromSequenceExclusive: cursor,
@@ -388,25 +378,21 @@ const makeEventStore = Effect.gen(function* () { · 14 unmodified lines
388378 ),
389379 ),
390380 ),
381+ Effect.map((events) => {
382+ const last = events.at(-1);
383+ const nextRemaining = remaining - events.length;
384+ return [
385+ events,
386+ last === undefined ||
387+ events.length < READ_PAGE_SIZE ||
388+ nextRemaining === 0 ||
389+ last.sequence >= input.toSequenceInclusive
390+ ? Option.none()
391+ : Option.some({ cursor: last.sequence, remaining: nextRemaining }),
392+ ] as const;
393+ }),
391394 ),
392− ).pipe(
393− Stream.flatMap((events) => {
394− const last = events.at(-1);
395− if (last === undefined) {
396− return Stream.empty;
397− }
398− const nextRemaining = remaining - events.length;
399− if (
400− events.length < READ_PAGE_SIZE ||
401− nextRemaining === 0 ||
402− last.sequence >= input.toSequenceInclusive
403− ) {
404− return Stream.fromIterable(events);
405− }
406− return Stream.concat(Stream.fromIterable(events), readPage(last.sequence, nextRemaining));
407− }),
408− );
409− return readPage(input.fromSequenceExclusive, limit);
395+ );
410396 };
411397
412398 const getAggregateReplayStats: OrchestrationEventStoreShape["getAggregateReplayStats"] = (