authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-10-10 13:48:02+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 06:48:02-05:00
log03b210528160e68c8dbfc27f11f693085f92ab09
tree62513709f743fb0792c37948608745b0ea28a036
parenta5b482452dc8ca87625d5a80f83be0a9b56674de
signature Signed by PGP key B5690EEEBB952194

fix(server): prevent busy terminals from starving history persistence (#17181)

2 files changed, 105 insertions(+), 29 deletions(-)

packages/shared/src/KeyedCoalescingWorker.test.ts+94-1
...@@ -6,6 +6,98 @@ import * as Effect from "effect/Effect";...@@ -6,6 +6,98 @@ import * as Effect from "effect/Effect";
6import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts";6import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts";
77
8describe("makeKeyedCoalescingWorker", () => {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
9 it.live("waits for latest work enqueued during active processing before draining the key", () =>101 it.live("waits for latest work enqueued during active processing before draining the key", () =>
10 Effect.scoped(102 Effect.scoped(
11 Effect.gen(function* () {103 Effect.gen(function* () {
...@@ -85,12 +177,13 @@ describe("makeKeyedCoalescingWorker", () => {...@@ -85,12 +177,13 @@ describe("makeKeyedCoalescingWorker", () => {
85177
86 yield* worker.enqueue("terminal-1", "first");178 yield* worker.enqueue("terminal-1", "first");
87 yield* Deferred.await(firstStarted);179 yield* Deferred.await(firstStarted);
180 yield* worker.enqueue("terminal-2", "quiet");
88 yield* worker.enqueue("terminal-1", "second");181 yield* worker.enqueue("terminal-1", "second");
89 yield* Deferred.succeed(releaseFailure, undefined);182 yield* Deferred.succeed(releaseFailure, undefined);
90 yield* Deferred.await(secondProcessed);183 yield* Deferred.await(secondProcessed);
91 yield* worker.drainKey("terminal-1");184 yield* worker.drainKey("terminal-1");
92185
93 expect(processed).toEqual(["terminal-1:first", "terminal-1:second"]);186 expect(processed).toEqual(["terminal-1:first", "terminal-2:quiet", "terminal-1:second"]);
94 }),187 }),
95 ),188 ),
96 );189 );
packages/shared/src/KeyedCoalescingWorker.ts+11-28
...@@ -2,8 +2,9 @@...@@ -2,8 +2,9 @@
2 * KeyedCoalescingWorker - A keyed worker that keeps only the latest value per key.2 * KeyedCoalescingWorker - A keyed worker that keeps only the latest value per key.
3 *3 *
4 * Enqueues for an active or already-queued key are merged atomically instead of4 * Enqueues for an active or already-queued key are merged atomically instead of
5 * creating duplicate queued items. `drainKey()` resolves only when that key has5 * creating duplicate queued items. Each batch yields to other queued keys before
6 * no queued, pending, or active work left.6 * processing newer updates. `drainKey()` resolves only when that key has no
7 * queued, pending, or active work left.
7 *8 *
8 * @module KeyedCoalescingWorker9 * @module KeyedCoalescingWorker
9 */10 */
...@@ -35,28 +36,9 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {...@@ -35,28 +36,9 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {
35 activeKeys: new Set(),36 activeKeys: new Set(),
36 });37 });
3738
38 const processKey = (key: K, value: V): Effect.Effect<void, E, R> =>39 // Return dirty keys to the tail on success or failure. Keep pending values
39 options.process(key, value).pipe(40 // available for merging until the next batch starts.
40 Effect.flatMap(() =>41 const completeKey = (key: K): Effect.Effect<void> =>
41 TxRef.modify(stateRef, (state) => {
42 const nextValue = state.latestByKey.get(key);
43 if (nextValue === undefined) {
44 const activeKeys = new Set(state.activeKeys);
45 activeKeys.delete(key);
46 return [null, { ...state, activeKeys }] as const;
47 }
48
49 const latestByKey = new Map(state.latestByKey);
50 latestByKey.delete(key);
51 return [nextValue, { ...state, latestByKey }] as const;
52 }).pipe(Effect.tx),
53 ),
54 Effect.flatMap((nextValue) =>
55 nextValue === null ? Effect.void : processKey(key, nextValue),
56 ),
57 );
58
59 const cleanupFailedKey = (key: K): Effect.Effect<void> =>
60 TxRef.modify(stateRef, (state) => {42 TxRef.modify(stateRef, (state) => {
61 const activeKeys = new Set(state.activeKeys);43 const activeKeys = new Set(state.activeKeys);
62 activeKeys.delete(key);44 activeKeys.delete(key);
...@@ -69,10 +51,11 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {...@@ -69,10 +51,11 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {
6951
70 return [false, { ...state, activeKeys }] as const;52 return [false, { ...state, activeKeys }] as const;
71 }).pipe(53 }).pipe(
72 Effect.tx,
73 Effect.flatMap((shouldRequeue) =>54 Effect.flatMap((shouldRequeue) =>
74 shouldRequeue ? TxQueue.offer(queue, key) : Effect.void,55 shouldRequeue ? TxQueue.offer(queue, key) : Effect.void,
75 ),56 ),
57 Effect.tx,
58 Effect.asVoid,
76 );59 );
7760
78 yield* TxQueue.take(queue).pipe(61 yield* TxQueue.take(queue).pipe(
...@@ -100,9 +83,9 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {...@@ -100,9 +83,9 @@ export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {
100 Effect.flatMap((item) =>83 Effect.flatMap((item) =>
101 item === null84 item === null
102 ? Effect.void85 ? Effect.void
103 : processKey(item.key, item.value).pipe(86 : options
104 Effect.catchCause(() => cleanupFailedKey(item.key)),87 .process(item.key, item.value)
105 ),88 .pipe(Effect.ignoreCause, Effect.ensuring(completeKey(item.key))),
106 ),89 ),
107 Effect.forever,90 Effect.forever,
108 Effect.forkScoped,91 Effect.forkScoped,