fix(shared): don't count dropped post-shutdown enqueues toward drain

packages/shared/src/DrainableWorker.test.ts

@@ -2,6 +2,8 @@ import { it } from "@effect/vitest";
 import { describe, expect } from "vite-plus/test";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Scope from "effect/Scope";
 
 import { makeDrainableWorker } from "./DrainableWorker.ts";
 
@@ -54,4 +56,22 @@ describe("makeDrainableWorker", () => {
       }),
     ),
   );
+
+  it.live("drain does not hang on work enqueued after the worker shut down", () =>
+    Effect.gen(function* () {
+      const scope = yield* Scope.make();
+      const worker = yield* makeDrainableWorker((_item: string) => Effect.void).pipe(
+        Scope.provide(scope),
+      );
+
+      yield* Scope.close(scope, Exit.void);
+
+      // The queue is shut down: the offer drops the item and reports false.
+      // The dropped item must not count toward drain, or drain hangs forever.
+      yield* worker.enqueue("late");
+
+      const drained = yield* worker.drain.pipe(Effect.timeout("2 seconds"), Effect.as("drained"));
+      expect(drained).toBe("drained");
+    }),
+  );
 });

packages/shared/src/DrainableWorker.ts

@@ -60,9 +60,12 @@ export const makeDrainableWorker = <A, E, R>(
       Effect.tx,
     );
 
-    const enqueue = (element: A): Effect.Effect<boolean, never, never> =>
+    const enqueue: DrainableWorker<A>["enqueue"] = (element: A) =>
       TxQueue.offer(queue, element).pipe(
-        Effect.tap(() => TxRef.update(outstanding, (n) => n + 1)),
+        // A shut-down queue drops the item and reports false; only track
+        // items that were actually accepted or drain() can hang forever.
+        Effect.tap((offered) => (offered ? TxRef.update(outstanding, (n) => n + 1) : Effect.void)),
+        Effect.asVoid,
         Effect.tx,
       );