authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 21:40:16-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 21:40:16-07:00
log24e4b62c883eb67251bdfb07d2bf22cfd3987820
tree22faff3aefb5689d75ffac04670312ba44744c2b
parent4b3568b775d86fc699938ec5a9ada773bad8b45f
signature Signed by PGP key B5690EEEBB952194

fix(shared): DrainableWorker keeps running after a failed item (#16223)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

2 files changed, 110 insertions(+), 9 deletions(-)

packages/shared/src/DrainableWorker.test.ts+77
......@@ -1,11 +1,88 @@
11import { it } from "@effect/vitest";
22import { describe, expect } from "vite-plus/test";
3import * as Cause from "effect/Cause";
34import * as Deferred from "effect/Deferred";
45import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
8import * as Logger from "effect/Logger";
9import * as Scope from "effect/Scope";
510
611import { makeDrainableWorker } from "./DrainableWorker.ts";
712
13const captureErrorLogs = () => {
14 const loggedErrors: Array<unknown> = [];
15 const logger = Logger.make(({ logLevel, cause }) => {
16 if (logLevel === "Error") loggedErrors.push(Cause.squash(cause));
17 });
18 return { loggedErrors, loggerLayer: Logger.layer([logger], { mergeWithExisting: false }) };
19};
20
821describe("makeDrainableWorker", () => {
22 it.effect("logs a failed or defective item and keeps processing later items", () => {
23 const { loggedErrors, loggerLayer } = captureErrorLogs();
24
25 return Effect.scoped(
26 Effect.gen(function* () {
27 const processed: string[] = [];
28 const workerFiber = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>();
29 const worker = yield* makeDrainableWorker((item: string) =>
30 Effect.gen(function* () {
31 yield* Deferred.succeed(workerFiber, yield* Effect.fiber);
32 if (item === "fail") return yield* Effect.fail("typed failure");
33 if (item === "die") return yield* Effect.die("defect");
34 if (item === "interrupt") return yield* Effect.interrupt;
35 processed.push(item);
36 }),
37 );
38
39 yield* worker.enqueue("fail");
40 yield* worker.enqueue("die");
41 yield* worker.enqueue("interrupt");
42 yield* worker.enqueue("ok");
43
44 // A stopped worker never drains; its exit settles the race instead of a hang.
45 yield* Effect.raceFirst(worker.drain, Fiber.await(yield* Deferred.await(workerFiber)));
46
47 expect(processed).toEqual(["ok"]);
48 // The item that interrupted itself was cancelled, not failed.
49 expect(loggedErrors).toEqual(["typed failure", "defect"]);
50 }),
51 ).pipe(Effect.provide(loggerLayer));
52 });
53
54 it.effect("stops quietly when its scope closes during an item", () => {
55 const { loggedErrors, loggerLayer } = captureErrorLogs();
56
57 return Effect.gen(function* () {
58 const scope = yield* Scope.make();
59 const processed: string[] = [];
60 const workerFiber = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>();
61 const worker = yield* makeDrainableWorker((item: string) =>
62 Effect.gen(function* () {
63 processed.push(item);
64 yield* Deferred.succeed(workerFiber, yield* Effect.fiber);
65 return yield* Effect.never;
66 }),
67 ).pipe(Scope.provide(scope));
68
69 yield* worker.enqueue("block");
70 yield* worker.enqueue("later");
71 const fiber = yield* Deferred.await(workerFiber);
72 yield* Scope.close(scope, Exit.void);
73
74 expect(Exit.hasInterrupts(yield* Fiber.await(fiber))).toBe(true);
75 expect(processed).toEqual(["block"]);
76 expect(loggedErrors).toEqual([]);
77
78 // Queued and late items are dropped with the queue, so drain resolves at once.
79 yield* worker.enqueue("after close");
80 const drained = yield* Effect.forkChild(worker.drain, { startImmediately: true });
81 expect(drained.pollUnsafe()).toEqual(Exit.void);
82 expect(processed).toEqual(["block"]);
83 }).pipe(Effect.provide(loggerLayer));
84 });
85
986 it.live("waits for work enqueued during active processing before draining", () =>
1087 Effect.scoped(
1188 Effect.gen(function* () {
packages/shared/src/DrainableWorker.ts+33-9
......@@ -8,6 +8,7 @@
88 *
99 * @module DrainableWorker
1010 */
11import * as Cause from "effect/Cause";
1112import * as Scope from "effect/Scope";
1213import * as Effect from "effect/Effect";
1314import * as TxQueue from "effect/TxQueue";
......@@ -32,23 +33,45 @@ export interface DrainableWorker<A> {
3233 * Create a drainable worker that processes items from an unbounded queue.
3334 *
3435 * The worker is forked into the current scope and will be interrupted when
35 * the scope closes. A finalizer shuts down the queue.
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.
3641 *
3742 * @param process - The effect to run for each queued item.
38 * @returns A `DrainableWorker` with `queue` and `drain`.
43 * @returns A `DrainableWorker` with `enqueue` and `drain`.
3944 */
4045export const makeDrainableWorker = <A, E, R>(
4146 process: (item: A) => Effect.Effect<void, E, R>,
4247): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>
4348 Effect.gen(function* () {
44 const queue = yield* Effect.acquireRelease(TxQueue.unbounded<A>(), TxQueue.shutdown);
4549 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 );
4659
4760 yield* TxQueue.take(queue).pipe(
48 Effect.tap((a) =>
49 Effect.ensuring(
50 process(a),
51 TxRef.update(outstanding, (n) => n - 1),
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)),
5275 ),
5376 ),
5477 Effect.forever,
......@@ -56,13 +79,14 @@ export const makeDrainableWorker = <A, E, R>(
5679 );
5780
5881 const drain: DrainableWorker<A>["drain"] = TxRef.get(outstanding).pipe(
59 Effect.tap((n) => (n > 0 ? Effect.txRetry : Effect.void)),
82 Effect.flatMap((n) => (n > 0 ? Effect.txRetry : Effect.void)),
6083 Effect.tx,
6184 );
6285
6386 const enqueue = (element: A): Effect.Effect<boolean, never, never> =>
6487 TxQueue.offer(queue, element).pipe(
65 Effect.tap(() => TxRef.update(outstanding, (n) => n + 1)),
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)),
6690 Effect.tx,
6791 );
6892