✨ Add configurable Worker shutdown policies

worker/README.md

@@ -14,11 +14,32 @@ thread.
 ## Features
 
 - Establishes two-way communication between the main and the worker threads
-- Gracefully shutdowns the worker from the main thread
+- Gracefully shuts down Workers by default with optional forceful termination
 - 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 a close message.
+Use the `terminate` policy when cancellation must be preemptible:
+
+```ts
+const worker = yield* useWorker("./worker.ts", {
+  type: "module",
+  shutdown: "terminate",
+});
+```
+
+When this Worker is still active during host teardown, `useWorker()` calls
+`Worker.terminate()` and reports that the Worker was terminated. This policy
+does not run Worker-side finalizers, so durable cleanup for terminated Workers
+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 configurable 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,8 +1,8 @@
 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,
   scoped,
@@ -83,7 +83,7 @@ describe("worker", () => {
       url = import.meta.resolve("./test-assets/shutdown-worker.ts");
     });
 
-    it("shuts down gracefully", function* () {
+    it("shuts down gracefully by default", function* () {
       let task = yield* spawn(function* () {
         yield* useWorker(url, {
           type: "module",
@@ -96,7 +96,6 @@ describe("worker", () => {
         yield* suspend();
       });
 
-      // Wait for worker to start
       yield* when(
         function* () {
           let exists = yield* until(
@@ -105,27 +104,51 @@ 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("terminates a CPU-bound worker", function* () {
+      expect.assertions(1);
+      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: "terminate" },
+        );
+        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}"`);
+          if (Atomics.load(state, 0) !== 1) {
+            throw new Error("worker has not started spinning");
           }
-          return text;
         },
-        { timeout: 500 },
+        { timeout: 10_000 },
       );
 
-      expect(content).toEqual("goodbye cruel world!");
+      yield* task.halt();
+
+      let taskError: Error | undefined;
+      try {
+        yield* task;
+      } catch (error) {
+        taskError = error as Error;
+      }
+      expect(taskError?.message).toContain("halted");
     });
   });
 

worker/worker.ts

@@ -69,6 +69,14 @@ export interface WorkerResource<TSend, TRecv, TReturn>
   ): Operation<TReturn>;
 }
 
+/** Options for creating and shutting down a Worker. */
+export interface UseWorkerOptions<TData> extends WorkerOptions {
+  /** Data passed to `workerMain()` during initialization. */
+  data?: TData;
+  /** How an active Worker shuts down with its host scope. Defaults to graceful. */
+  shutdown?: "graceful" | "terminate";
+}
+
 /**
  * Use on the main thread to create and exeecute a well behaved web worker.
  *
@@ -110,7 +118,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,7 +127,7 @@ 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 outcome = withResolvers<TReturn>();
@@ -217,23 +225,12 @@ 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 (options?.shutdown === "terminate") {
+          worker.terminate();
+          rejectOutcome(new Error("worker terminated"));
+        } else {
+          worker.postMessage({ type: "close" });
         }
       }
       yield* settled(outcome.operation);