Rename keyed worker and update terminal manager

- Rename the shared coalescing worker API and exports
- Switch terminal persistence to the keyed worker
- Drop the unused global drain helper

Author: juliusmarminge

apps/server/src/terminal/Layers/Manager.ts

@@ -6,7 +6,7 @@ import {
   type TerminalSessionSnapshot,
   type TerminalSessionStatus,
 } from "@t3tools/contracts";
-import { makeCoalescingDrainableWorker } from "@t3tools/shared/CoalescingDrainableWorker";
+import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker";
 import {
   Data,
   Effect,
@@ -804,7 +804,7 @@ export const makeTerminalManagerWithOptions = Effect.fn("makeTerminalManagerWith
       yield* registerKillFiber(process, fiber);
     });
 
-    const persistWorker = yield* makeCoalescingDrainableWorker<
+    const persistWorker = yield* makeKeyedCoalescingWorker<
       string,
       PersistHistoryRequest,
       never,

packages/shared/package.json

@@ -28,9 +28,9 @@
       "types": "./src/DrainableWorker.ts",
       "import": "./src/DrainableWorker.ts"
     },
-    "./CoalescingDrainableWorker": {
-      "types": "./src/CoalescingDrainableWorker.ts",
-      "import": "./src/CoalescingDrainableWorker.ts"
+    "./KeyedCoalescingWorker": {
+      "types": "./src/KeyedCoalescingWorker.ts",
+      "import": "./src/KeyedCoalescingWorker.ts"
     },
     "./schemaJson": {
       "types": "./src/schemaJson.ts",

…ed/src/CoalescingDrainableWorker.test.ts …shared/src/KeyedCoalescingWorker.test.tspackages/shared/src/CoalescingDrainableWorker.test.ts renamed to packages/shared/src/KeyedCoalescingWorker.test.ts packages/shared/src/CoalescingDrainableWorker.test.ts renamed to packages/shared/src/KeyedCoalescingWorker.test.ts

@@ -2,9 +2,9 @@ import { it } from "@effect/vitest";
 import { describe, expect } from "vitest";
 import { Deferred, Effect } from "effect";
 
-import { makeCoalescingDrainableWorker } from "./CoalescingDrainableWorker";
+import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker";
 
-describe("makeCoalescingDrainableWorker", () => {
+describe("makeKeyedCoalescingWorker", () => {
   it.live("waits for latest work enqueued during active processing before draining the key", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -14,7 +14,7 @@ describe("makeCoalescingDrainableWorker", () => {
         const secondStarted = yield* Deferred.make<void>();
         const releaseSecond = yield* Deferred.make<void>();
 
-        const worker = yield* makeCoalescingDrainableWorker<string, string, never, never>({
+        const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
           merge: (_current, next) => next,
           process: (key, value) =>
             Effect.gen(function* () {
@@ -64,7 +64,7 @@ describe("makeCoalescingDrainableWorker", () => {
         const releaseFailure = yield* Deferred.make<void>();
         const secondProcessed = yield* Deferred.make<void>();
 
-        const worker = yield* makeCoalescingDrainableWorker<string, string, string, never>({
+        const worker = yield* makeKeyedCoalescingWorker<string, string, string, never>({
           merge: (_current, next) => next,
           process: (key, value) =>
             Effect.gen(function* () {

…/shared/src/CoalescingDrainableWorker.ts …ages/shared/src/KeyedCoalescingWorker.tspackages/shared/src/CoalescingDrainableWorker.ts renamed to packages/shared/src/KeyedCoalescingWorker.ts packages/shared/src/CoalescingDrainableWorker.ts renamed to packages/shared/src/KeyedCoalescingWorker.ts

@@ -1,34 +1,33 @@
 /**
- * CoalescingDrainableWorker - A keyed worker that keeps only the latest value per key.
+ * KeyedCoalescingWorker - A keyed worker that keeps only the latest value per key.
  *
  * Enqueues for an active or already-queued key are merged atomically instead of
  * creating duplicate queued items. `drainKey()` resolves only when that key has
  * no queued, pending, or active work left.
  *
- * @module CoalescingDrainableWorker
+ * @module KeyedCoalescingWorker
  */
 import type { Scope } from "effect";
 import { Effect, TxQueue, TxRef } from "effect";
 
-export interface CoalescingDrainableWorker<K, V> {
+export interface KeyedCoalescingWorker<K, V> {
   readonly enqueue: (key: K, value: V) => Effect.Effect<void>;
   readonly drainKey: (key: K) => Effect.Effect<void>;
-  readonly drain: Effect.Effect<void>;
 }
 
-interface CoalescingWorkerState<K, V> {
+interface KeyedCoalescingWorkerState<K, V> {
   readonly latestByKey: Map<K, V>;
   readonly queuedKeys: Set<K>;
   readonly activeKeys: Set<K>;
 }
 
-export const makeCoalescingDrainableWorker = <K, V, E, R>(options: {
+export const makeKeyedCoalescingWorker = <K, V, E, R>(options: {
   readonly merge: (current: V, next: V) => V;
   readonly process: (key: K, value: V) => Effect.Effect<void, E, R>;
-}): Effect.Effect<CoalescingDrainableWorker<K, V>, never, Scope.Scope | R> =>
+}): Effect.Effect<KeyedCoalescingWorker<K, V>, never, Scope.Scope | R> =>
   Effect.gen(function* () {
     const queue = yield* Effect.acquireRelease(TxQueue.unbounded<K>(), TxQueue.shutdown);
-    const stateRef = yield* TxRef.make<CoalescingWorkerState<K, V>>({
+    const stateRef = yield* TxRef.make<KeyedCoalescingWorkerState<K, V>>({
       latestByKey: new Map(),
       queuedKeys: new Set(),
       activeKeys: new Set(),
@@ -107,7 +106,7 @@ export const makeCoalescingDrainableWorker = <K, V, E, R>(options: {
       Effect.forkScoped,
     );
 
-    const enqueue: CoalescingDrainableWorker<K, V>["enqueue"] = (key, value) =>
+    const enqueue: KeyedCoalescingWorker<K, V>["enqueue"] = (key, value) =>
       TxRef.modify(stateRef, (state) => {
         const latestByKey = new Map(state.latestByKey);
         const existing = latestByKey.get(key);
@@ -126,17 +125,7 @@ export const makeCoalescingDrainableWorker = <K, V, E, R>(options: {
         Effect.asVoid,
       );
 
-    const drain: CoalescingDrainableWorker<K, V>["drain"] = TxRef.get(stateRef).pipe(
-      Effect.tap((state) =>
-        state.latestByKey.size > 0 || state.queuedKeys.size > 0 || state.activeKeys.size > 0
-          ? Effect.txRetry
-          : Effect.void,
-      ),
-      Effect.asVoid,
-      Effect.tx,
-    );
-
-    const drainKey: CoalescingDrainableWorker<K, V>["drainKey"] = (key) =>
+    const drainKey: KeyedCoalescingWorker<K, V>["drainKey"] = (key) =>
       TxRef.get(stateRef).pipe(
         Effect.tap((state) =>
           state.latestByKey.has(key) || state.queuedKeys.has(key) || state.activeKeys.has(key)
@@ -147,5 +136,5 @@ export const makeCoalescingDrainableWorker = <K, V, E, R>(options: {
         Effect.tx,
       );
 
-    return { enqueue, drainKey, drain } satisfies CoalescingDrainableWorker<K, V>;
+    return { enqueue, drainKey } satisfies KeyedCoalescingWorker<K, V>;
   });