✨ Add configurable Worker shutdown policies

worker/README.md

@@ -14,11 +14,61 @@ thread.
 ## Features
 
 - Establishes two-way communication between the main and the worker threads
-- Gracefully shutdowns the worker from the main thread
+- Gracefully or forcibly shuts down Workers with contextual policies
 - Propagates errors from the worker to the main thread
 - Type-safe message handling with TypeScript
 - Supports worker-initiated requests handled by the host
 
+## Shutdown
+
+Workers shut down gracefully by default. When their host scope shuts down,
+`useWorker()` sends a close message and waits for Worker-side teardown and the
+final result.
+
+CPU-bound or otherwise non-cooperative Workers cannot process that close
+message. Use `shutdown: "forced"` when cancellation should terminate the Worker
+immediately:
+
+```ts
+const worker = yield* useWorker("./worker.ts", {
+  type: "module",
+  shutdown: "forced",
+});
+```
+
+A generator policy first requests graceful shutdown, then chooses whether to
+keep waiting or force termination. It runs in the Worker's host evaluation
+scope, so it can use application context. For example, a host-owned heartbeat
+monitor can expose the point when the Worker's control channel stops responding:
+
+```ts
+import { createContext, type Operation } from "effection";
+
+interface WorkerHealth {
+  controlChannelUnresponsive: Operation<void>;
+}
+
+const workerHealth = createContext<WorkerHealth>("worker health");
+
+const worker = yield* useWorker("./worker.ts", {
+  type: "module",
+  *shutdown() {
+    const health = yield* workerHealth.expect();
+    yield* health.controlChannelUnresponsive;
+    return "forced";
+  },
+});
+```
+
+The heartbeat monitor is application-owned because the package cannot infer
+whether CPU use or delayed messages mean that a particular Worker is unhealthy.
+If the Worker completes while the policy is pending, Effection cancels the
+policy and finishes gracefully. Returning `"graceful"` also keeps waiting for
+Worker-side teardown. No timeout is imposed by the package.
+
+Forced termination does not run Worker-side finalizers, so durable cleanup for
+a forcibly terminated Worker must be owned by the host.
+
 ## Usage: Get worker's return value
 
 The return value of the worker is the return value of the function passed to

worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/worker",
-  "description": "Web Worker integration with two-way messaging and graceful shutdown",
-  "version": "0.5.4",
+  "description": "Web Worker integration with two-way messaging and policy-driven shutdown",
+  "version": "0.6.0",
   "keywords": ["platform"],
   "type": "module",
   "main": "./dist/mod.js",

worker/test-assets/cpu-bound-worker.ts

@@ -0,0 +1,11 @@
+import { workerMain } from "../worker-main.ts";
+
+await workerMain<never, never, never, SharedArrayBuffer>(function* ({ data }) {
+  let state = new Int32Array(data);
+  Atomics.store(state, 0, 1);
+  Atomics.notify(state, 0);
+
+  while (true) {
+    Atomics.load(state, 0);
+  }
+});

worker/worker.test.ts

@@ -1,10 +1,12 @@
 import { access, mkdir, readFile, rm } from "node:fs/promises";
 import { join } from "node:path";
 import { fileURLToPath } from "node:url";
-import { beforeEach, describe, it } from "@effectionx/vitest";
 import { when } from "@effectionx/converge";
+import { beforeEach, describe, it } from "@effectionx/vitest";
 import {
   all,
+  createContext,
+  type Operation,
   scoped,
   sleep,
   spawn,
@@ -15,7 +17,11 @@ import {
 import { expect } from "expect";
 
 import type { ShutdownWorkerParams } from "./test-assets/shutdown-worker.ts";
-import { useWorker } from "./worker.ts";
+import {
+  type UseWorkerOptions,
+  type WorkerShutdownPolicy,
+  useWorker,
+} from "./worker.ts";
 
 describe("worker", () => {
   it("sends and receive messages in synchrony", function* () {
@@ -83,7 +89,43 @@ describe("worker", () => {
       url = import.meta.resolve("./test-assets/shutdown-worker.ts");
     });
 
-    it("shuts down gracefully", function* () {
+    function* haltCPUWorker(
+      shutdown: UseWorkerOptions<SharedArrayBuffer>["shutdown"],
+    ): Operation<Error | undefined> {
+      let state = new Int32Array(
+        new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT),
+      );
+      let task = yield* spawn(function* () {
+        yield* useWorker(
+          import.meta.resolve("./test-assets/cpu-bound-worker.ts"),
+          {
+            type: "module",
+            data: state.buffer,
+            shutdown,
+          },
+        );
+        yield* suspend();
+      });
+
+      yield* when(
+        function* () {
+          if (Atomics.load(state, 0) !== 1) {
+            throw new Error("worker has not started spinning");
+          }
+        },
+        { timeout: 10_000 },
+      );
+
+      yield* task.halt();
+
+      try {
+        yield* task;
+      } catch (error) {
+        return error as Error;
+      }
+    }
+
+    it("shuts down gracefully by default", function* () {
       let task = yield* spawn(function* () {
         yield* useWorker(url, {
           type: "module",
@@ -96,7 +138,6 @@ describe("worker", () => {
         yield* suspend();
       });
 
-      // Wait for worker to start
       yield* when(
         function* () {
           let exists = yield* until(
@@ -105,27 +146,95 @@ describe("worker", () => {
               () => false,
             ),
           );
-          if (!exists) throw new Error("start file not found");
-          return true;
+          if (!exists) {
+            throw new Error("worker has not started");
+          }
         },
         { timeout: 10_000 },
       );
 
       yield* task.halt();
 
-      // Wait for the end file to be written with expected content
-      let { value: content } = yield* when(
+      expect(yield* until(readFile(endFile, "utf-8"))).toEqual(
+        "goodbye cruel world!",
+      );
+    });
+
+    it("cancels its shutdown policy when graceful shutdown completes", function* () {
+      let shutdownContext = createContext<string>("worker shutdown test");
+      yield* shutdownContext.set("available during shutdown");
+
+      let policyContext: string | undefined;
+      let terminated = false;
+      let task = yield* spawn(function* () {
+        yield* useWorker(url, {
+          type: "module",
+          data: {
+            startFile,
+            endFile,
+            endText: "graceful",
+          } satisfies ShutdownWorkerParams,
+          *shutdown() {
+            policyContext = yield* shutdownContext.expect();
+            yield* suspend();
+            terminated = true;
+            return "forced";
+          },
+        });
+        yield* suspend();
+      });
+
+      yield* when(
         function* () {
-          let text = yield* until(readFile(endFile, "utf-8").catch(() => ""));
-          if (text !== "goodbye cruel world!") {
-            throw new Error(`expected "goodbye cruel world!", got "${text}"`);
+          let exists = yield* until(
+            access(startFile).then(
+              () => true,
+              () => false,
+            ),
+          );
+          if (!exists) {
+            throw new Error("worker has not started");
           }
-          return text;
         },
-        { timeout: 500 },
+        { timeout: 10_000 },
       );
 
-      expect(content).toEqual("goodbye cruel world!");
+      yield* task.halt();
+
+      expect(yield* until(readFile(endFile, "utf-8"))).toEqual("graceful");
+      expect(policyContext).toEqual("available during shutdown");
+      expect(terminated).toEqual(false);
+    });
+
+    it("terminates a CPU-bound worker in forced mode", function* () {
+      expect.assertions(1);
+      const taskError = yield* haltCPUWorker("forced");
+      expect(taskError?.message).toContain("halted");
+    });
+
+    it("can force a CPU-bound worker from host health state", function* () {
+      expect.assertions(2);
+      const workerHealth = createContext<{
+        controlChannelUnresponsive: Operation<void>;
+      }>("worker health");
+      const controlChannelUnresponsive = withResolvers<void>();
+      yield* workerHealth.set({
+        controlChannelUnresponsive: controlChannelUnresponsive.operation,
+      });
+      controlChannelUnresponsive.resolve();
+
+      let observedHealth = false;
+      const shutdown: WorkerShutdownPolicy = function* () {
+        const health = yield* workerHealth.expect();
+        observedHealth = true;
+        yield* health.controlChannelUnresponsive;
+        return "forced";
+      };
+
+      const taskError = yield* haltCPUWorker(shutdown);
+
+      expect(observedHealth).toEqual(true);
+      expect(taskError?.message).toContain("halted");
     });
   });
 

worker/worker.ts

@@ -7,6 +7,7 @@ import {
   ensure,
   on,
   once,
+  race,
   resource,
   spawn,
   withResolvers,
@@ -69,6 +70,20 @@ export interface WorkerResource<TSend, TRecv, TReturn>
   ): Operation<TReturn>;
 }
 
+/** How an active resource should shut down with its owning scope. */
+export type ShutdownMode = "graceful" | "forced";
+
+/** Selects a Worker shutdown mode from application state. */
+export type WorkerShutdownPolicy = () => Operation<ShutdownMode>;
+
+/** Options for creating and shutting down a Worker. */
+export interface UseWorkerOptions<TData> extends WorkerOptions {
+  /** Data passed to `workerMain()` during initialization. */
+  data?: TData;
+  /** Selects graceful, forced, or policy-driven shutdown. */
+  shutdown?: ShutdownMode | WorkerShutdownPolicy;
+}
+
 /**
  * Use on the main thread to create and exeecute a well behaved web worker.
  *
@@ -110,7 +125,7 @@ export interface WorkerResource<TSend, TRecv, TReturn>
  * ```
  *
  * @param url URL or string of script
- * @param options WorkerOptions
+ * @param options Worker construction and shutdown options
  * @template TSend - value main thread will send to the worker
  * @template TRecv - value main thread will receive from the worker
  * @template TReturn - worker operation return value
@@ -119,9 +134,10 @@ export interface WorkerResource<TSend, TRecv, TReturn>
  */
 export function useWorker<TSend, TRecv, TReturn, TData>(
   url: string | URL,
-  options?: WorkerOptions & { data?: TData },
+  options?: UseWorkerOptions<TData>,
 ): Operation<WorkerResource<TSend, TRecv, TReturn>> {
   return resource(function* (provide) {
+    let { data, shutdown = "graceful", ...workerOptions } = options ?? {};
     let outcome = withResolvers<TReturn>();
     let outcomeSettled = false;
 
@@ -141,7 +157,14 @@ export function useWorker<TSend, TRecv, TReturn, TData>(
       outcome.reject(error);
     };
 
-    let worker = new Worker(url, options);
+    let worker = new Worker(url, workerOptions);
+    const terminate = (error = new Error("worker terminated")) => {
+      if (!outcomeSettled) {
+        worker.terminate();
+        rejectOutcome(error);
+      }
+    };
+
     let subscription = yield* on(worker, "message");
 
     // Channel for worker-initiated requests (buffered via eager subscription)
@@ -217,21 +240,25 @@ export function useWorker<TSend, TRecv, TReturn, TData>(
     });
 
     yield* ensure(function* () {
-      worker.postMessage({ type: "close" });
       if (!outcomeSettled) {
-        while (!outcomeSettled) {
-          const event = yield* once(worker, "message");
-          const msg = event.data;
-          if (msg.type === "close") {
-            const { result } = msg as { result: Result<TReturn> };
-            if (result.ok) {
-              resolveOutcome(result.value);
-            } else {
-              const serializedError =
-                result.error as unknown as SerializedError;
-              rejectOutcome(
-                errorFromSerialized("Worker failed", serializedError),
-              );
+        if (shutdown === "forced") {
+          terminate();
+        } else {
+          worker.postMessage({ type: "close" });
+          if (typeof shutdown === "function") {
+            let mode: ShutdownMode = "graceful";
+            try {
+              mode = yield* race([
+                gracefulCompletion(outcome.operation),
+                shutdown(),
+              ]);
+            } catch (error) {
+              if (!outcomeSettled) {
+                terminate(error as Error);
+              }
+            }
+            if (mode === "forced") {
+              terminate();
             }
           }
         }
@@ -241,7 +268,7 @@ export function useWorker<TSend, TRecv, TReturn, TData>(
 
     worker.postMessage({
       type: "init",
-      data: options?.data,
+      data,
     });
 
     yield* provide({
@@ -329,6 +356,13 @@ export function useWorker<TSend, TRecv, TReturn, TData>(
   });
 }
 
+function* gracefulCompletion<T>(
+  outcome: Operation<T>,
+): Operation<ShutdownMode> {
+  yield* outcome;
+  return "graceful";
+}
+
 function settled<T>(operation: Operation<T>): Operation<Result<void>> {
   return {
     *[Symbol.iterator]() {