✨ Add configurable Worker shutdown policies

worker/README.md

@@ -14,11 +14,59 @@ 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 with contextual termination 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. Install shutdown middleware when cancellation must eventually become
+preemptible:
+
+```ts
+import { sleep } from "effection";
+
+const worker = yield* useWorker("./worker.ts", {
+  type: "module",
+  *shutdown(args, terminate) {
+    yield* sleep(2_000);
+    return yield* terminate(...args);
+  },
+});
+```
+
+The close message is posted before middleware runs. Middleware is raced against
+the Worker's result, so a Worker that completes gracefully during the delay
+cancels the pending escalation. Calling `terminate()`—the middleware chain's
+`next()` operation—hard-terminates the Worker only if it is still active. A
+middleware that returns without calling `terminate()` leaves shutdown graceful.
+
+Shutdown middleware is an Effection operation and can read context or other
+application state when selecting a policy:
+
+```ts
+const worker = yield* useWorker("./worker.ts", { type: "module" });
+
+yield* worker.around({
+  *shutdown(args, terminate) {
+    let usage = yield* measureCPUUsage();
+    yield* sleep(usage < 0.5 ? 10_000 : usage < 0.9 ? 2_000 : 100);
+    return yield* terminate(...args);
+  },
+});
+```
+
+Without shutdown middleware, `useWorker()` preserves the existing behavior: it
+waits for graceful Worker-side teardown without imposing a deadline. Hard
+termination does not run Worker-side finalizers, so durable cleanup for a
+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",
@@ -28,6 +28,8 @@
   },
   "sideEffects": false,
   "dependencies": {
+    "@effectionx/context-api": "workspace:*",
+    "@effectionx/scope-eval": "workspace:*",
     "@effectionx/signals": "workspace:*",
     "@effectionx/timebox": "workspace:*",
     "web-worker": "^1"

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/tsconfig.json

@@ -7,9 +7,15 @@
   "include": ["**/*.ts"],
   "exclude": ["**/*.test.ts", "test-assets/**", "dist"],
   "references": [
+    {
+      "path": "../context-api"
+    },
     {
       "path": "../converge"
     },
+    {
+      "path": "../scope-eval"
+    },
     {
       "path": "../signals"
     },

worker/worker.test.ts

@@ -1,10 +1,11 @@
 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,
   scoped,
   sleep,
   spawn,
@@ -83,7 +84,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 +97,6 @@ describe("worker", () => {
         yield* suspend();
       });
 
-      // Wait for worker to start
       yield* when(
         function* () {
           let exists = yield* until(
@@ -105,27 +105,105 @@ 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* () {
+        let worker = yield* useWorker(url, {
+          type: "module",
+          data: {
+            startFile,
+            endFile,
+            endText: "graceful",
+          } satisfies ShutdownWorkerParams,
+        });
+        yield* worker.around({
+          *shutdown(_args, terminate) {
+            policyContext = yield* shutdownContext.expect();
+            yield* suspend();
+            terminated = true;
+            return yield* 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}"`);
+          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 },
+      );
+
+      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", 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(args, terminate) {
+              return yield* terminate(...args);
+            },
+          },
+        );
+        yield* suspend();
+      });
 
-      expect(content).toEqual("goodbye cruel world!");
+      yield* when(
+        function* () {
+          if (Atomics.load(state, 0) !== 1) {
+            throw new Error("worker has not started spinning");
+          }
+        },
+        { timeout: 10_000 },
+      );
+
+      yield* task.halt();
+
+      let taskError: Error | undefined;
+      try {
+        yield* task;
+      } catch (error) {
+        taskError = error as Error;
+      }
+      expect(taskError?.message).toContain("halted");
     });
   });