1/**
2 * DrainableWorker - A queue-based worker that exposes a `drain()` effect.
3 *
4 * Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds
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()`.
8 *
9 * @module DrainableWorker
10 */
11import * as Cause from "effect/Cause";
12import * as Scope from "effect/Scope";
13import * as Effect from "effect/Effect";
14import * as TxQueue from "effect/TxQueue";
15import * as TxRef from "effect/TxRef";
16
17export interface DrainableWorker<A> {
18 /**
19 * Enqueue a work item and track it for `drain()`.
20 *
21 * This wraps `Queue.offer` so drain state is updated atomically with the
22 * enqueue path instead of inferring it from queue internals.
23 */
24 readonly enqueue: (item: A) => Effect.Effect<void>;
25
26 /**
27 * Resolves when the queue is empty and the worker is idle (not processing).
28 */
29 readonly drain: Effect.Effect<void>;
30}
31
32/**
33 * Create a drainable worker that processes items from an unbounded queue.
34 *
35 * The worker is forked into the current scope and will be interrupted when
36 * the scope closes. A finalizer shuts down the queue and drops queued items,
37 * so `drain` resolves after the scope closes instead of waiting on them.
38 *
39 * An item that fails or dies is logged and skipped; the worker keeps
40 * processing later items and `drain` still resolves.
41 *
42 * @param process - The effect to run for each queued item.
43 * @returns A `DrainableWorker` with `enqueue` and `drain`.
44 */
45export const makeDrainableWorker = <A, E, R>(
46 process: (item: A) => Effect.Effect<void, E, R>,
47): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>
48 Effect.gen(function* () {
49 const outstanding = yield* TxRef.make(0);
50 const queue = yield* Effect.acquireRelease(TxQueue.unbounded<A>(), (queue) =>
51 // Uncount only the dropped items: an item still running uncounts itself,
52 // even when a parallel scope closes it after this finalizer.
53 TxQueue.clear(queue).pipe(
54 Effect.flatMap((dropped) => TxRef.update(outstanding, (n) => n - dropped.length)),
55 Effect.andThen(TxQueue.shutdown(queue)),
56 Effect.tx,
57 ),
58 );
59
60 yield* TxQueue.take(queue).pipe(
61 Effect.flatMap((a) =>
62 // `suspend` turns a `process` that throws while building its effect
63 // into this item's defect instead of the loop's.
64 Effect.suspend(() => process(a)).pipe(
65 // Only the item's own failure, defect, or interruption lands here and
66 // the loop continues; interrupting the worker fiber still stops it.
67 // Callers treat an item that only interrupted itself as cancelled,
68 // not failed.
69 Effect.catchCause((cause) =>
70 Cause.hasInterruptsOnly(cause)
71 ? Effect.void
72 : Effect.logError("DrainableWorker item failed", cause),
73 ),
74 Effect.ensuring(TxRef.update(outstanding, (n) => n - 1)),
75 ),
76 ),
77 Effect.forever,
78 Effect.forkScoped,
79 );
80
81 const drain: DrainableWorker<A>["drain"] = TxRef.get(outstanding).pipe(
82 Effect.flatMap((n) => (n > 0 ? Effect.txRetry : Effect.void)),
83 Effect.tx,
84 );
85
86 const enqueue = (element: A): Effect.Effect<boolean, never, never> =>
87 TxQueue.offer(queue, element).pipe(
88 // A shut-down queue refuses the item, so it is never processed.
89 Effect.tap((offered) => (offered ? TxRef.update(outstanding, (n) => n + 1) : Effect.void)),
90 Effect.tx,
91 );
92
93 return { enqueue, drain } satisfies DrainableWorker<A>;
94 });