authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-28 22:52:39-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-28 22:52:39-07:00
log00f3efb95b485a40380af8295a5119b03095858b
tree7ead71298554910a6be14c691ee52e58732af391
parent7f3387a86cf6ed2e9bb7318db56b1ab19de80e69
signature Signed by PGP key B5690EEEBB952194

Make DrainableWorker drain off outstanding count (#1514)

1 files changed, 24 insertions(+), 32 deletions(-)

packages/shared/src/DrainableWorker.ts+24-32
...@@ -1,13 +1,15 @@...@@ -1,13 +1,15 @@
1/**1/**
2 * DrainableWorker - A queue-based worker with deterministic `drain()`.2 * DrainableWorker - A queue-based worker that exposes a `drain()` effect.
3 *3 *
4 * Tracks outstanding work in STM so `drain()` resolves only when no items4 * Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds
5 * are queued or in flight. Useful in tests instead of timing-based waits.5 * a signal that resolves when the queue is empty **and** the current item
6 * has finished processing. This lets tests replace timing-sensitive
7 * `Effect.sleep` calls with deterministic `drain()`.
6 *8 *
7 * @module DrainableWorker9 * @module DrainableWorker
8 */10 */
9import { Effect, TxQueue, TxRef } from "effect";
10import type { Scope } from "effect";11import type { Scope } from "effect";
12import { Effect, TxQueue, TxRef } from "effect";
1113
12export interface DrainableWorker<A> {14export interface DrainableWorker<A> {
13 /**15 /**
...@@ -37,40 +39,30 @@ export const makeDrainableWorker = <A, E, R>(...@@ -37,40 +39,30 @@ export const makeDrainableWorker = <A, E, R>(
37 process: (item: A) => Effect.Effect<void, E, R>,39 process: (item: A) => Effect.Effect<void, E, R>,
38): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>40): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>
39 Effect.gen(function* () {41 Effect.gen(function* () {
40 const ref = yield* TxRef.make(0);42 const queue = yield* Effect.acquireRelease(TxQueue.unbounded<A>(), TxQueue.shutdown);
4143 const outstanding = yield* TxRef.make(0);
42 const queue = yield* Effect.acquireRelease(TxQueue.unbounded<A>(), (queue) =>
43 TxQueue.shutdown(queue),
44 );
4544
46 const takeItem = Effect.tx(45 yield* TxQueue.take(queue).pipe(
47 Effect.gen(function* () {46 Effect.tap((a) =>
48 const item = yield* TxQueue.take(queue);47 Effect.ensuring(
49 yield* TxRef.update(ref, (n) => n + 1);48 process(a),
50 return item;49 TxRef.update(outstanding, (n) => n - 1),
51 }),50 ),
52 );
53
54 yield* takeItem.pipe(
55 Effect.flatMap((item) =>
56 process(item).pipe(Effect.ensuring(TxRef.update(ref, (n) => n - 1))),
57 ),51 ),
58 Effect.forever,52 Effect.forever,
59 Effect.forkScoped,53 Effect.forkScoped,
60 );54 );
6155
62 const drain: DrainableWorker<A>["drain"] = Effect.tx(56 const drain: DrainableWorker<A>["drain"] = TxRef.get(outstanding).pipe(
63 Effect.gen(function* () {57 Effect.tap((n) => (n > 0 ? Effect.txRetry : Effect.void)),
64 const inFlight = yield* TxRef.get(ref);58 Effect.tx,
65 const isEmpty = yield* TxQueue.isEmpty(queue);
66 if (inFlight > 0 || !isEmpty) {
67 return yield* Effect.txRetry;
68 }
69 }),
70 );59 );
7160
72 return {61 const enqueue = (element: A): Effect.Effect<boolean, never, never> =>
73 enqueue: (item) => TxQueue.offer(queue, item),62 TxQueue.offer(queue, element).pipe(
74 drain,63 Effect.tap(() => TxRef.update(outstanding, (n) => n + 1)),
75 } satisfies DrainableWorker<A>;64 Effect.tx,
65 );
66
67 return { enqueue, drain } satisfies DrainableWorker<A>;
76 });68 });