From 9ca7c9802eb69b9e887fc76f82f55be114cf5ce7 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 09:54:13 -0400 Subject: [PATCH 01/12] =?UTF-8?q?=F0=9F=90=9B=20Terminate=20active=20worke?= =?UTF-8?q?rs=20during=20teardown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- worker/README.md | 14 +++++- worker/package.json | 4 +- worker/test-assets/cpu-bound-worker.ts | 11 ++++ worker/worker.test.ts | 70 +++++++------------------- worker/worker.ts | 19 +------ 5 files changed, 47 insertions(+), 71 deletions(-) create mode 100644 worker/test-assets/cpu-bound-worker.ts diff --git a/worker/README.md b/worker/README.md index 4b309e22..f42f8949 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,11 +14,23 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Gracefully shutdowns the worker from the main thread +- Preemptibly terminates active workers when their host scope shuts down - 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 that finish on their own deliver their result normally. When the host +scope shuts down while a Worker is still active, `useWorker()` calls +`Worker.terminate()`. Cancellation therefore does not depend on the Worker's +event loop, and can reclaim Workers that are CPU-bound or otherwise +non-cooperative. Worker-side teardown is not guaranteed during cancellation; +durable cleanup should remain owned by the host. + +Applications upgrading from `0.5` that relied on Worker-side finalizers during +host cancellation should move that cleanup to the host before upgrading. + ## Usage: Get worker's return value The return value of the worker is the return value of the function passed to diff --git a/worker/package.json b/worker/package.json index b0ecce5c..3e4bb4a2 100644 --- a/worker/package.json +++ b/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 preemptible cancellation", + "version": "0.6.0", "keywords": ["platform"], "type": "module", "main": "./dist/mod.js", diff --git a/worker/test-assets/cpu-bound-worker.ts b/worker/test-assets/cpu-bound-worker.ts new file mode 100644 index 00000000..6b97e83f --- /dev/null +++ b/worker/test-assets/cpu-bound-worker.ts @@ -0,0 +1,11 @@ +import { workerMain } from "../worker-main.ts"; + +await workerMain(function* ({ data }) { + let state = new Int32Array(data); + Atomics.store(state, 0, 1); + Atomics.notify(state, 0); + + while (true) { + Atomics.load(state, 0); + } +}); diff --git a/worker/worker.test.ts b/worker/worker.test.ts index 9814b0ff..25a1383f 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -1,8 +1,5 @@ -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 { describe, it } from "@effectionx/vitest"; import { all, scoped, @@ -14,7 +11,6 @@ import { } from "effection"; import { expect } from "expect"; -import type { ShutdownWorkerParams } from "./test-assets/shutdown-worker.ts"; import { useWorker } from "./worker.ts"; describe("worker", () => { @@ -67,65 +63,37 @@ describe("worker", () => { } }); describe("shutdown", () => { - let startFile: string; - let endFile: string; - let url: string; - - beforeEach(function* () { - let dir = fileURLToPath(import.meta.resolve("./test-tmp")); - yield* until( - rm(dir, { recursive: true, force: true }).then(() => - mkdir(dir, { recursive: true }), - ), + it("terminates a CPU-bound worker", function* () { + expect.assertions(1); + let state = new Int32Array( + new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT), ); - startFile = join(dir, "started.txt"); - endFile = join(dir, "ended.txt"); - url = import.meta.resolve("./test-assets/shutdown-worker.ts"); - }); - - it("shuts down gracefully", function* () { let task = yield* spawn(function* () { - yield* useWorker(url, { - type: "module", - data: { - startFile, - endFile, - endText: "goodbye cruel world!", - } satisfies ShutdownWorkerParams, - }); + yield* useWorker( + import.meta.resolve("./test-assets/cpu-bound-worker.ts"), + { type: "module", data: state.buffer }, + ); yield* suspend(); }); - // Wait for worker to start yield* when( function* () { - let exists = yield* until( - access(startFile).then( - () => true, - () => false, - ), - ); - if (!exists) throw new Error("start file not found"); - return true; + if (Atomics.load(state, 0) !== 1) { + throw new Error("worker has not started spinning"); + } }, { timeout: 10_000 }, ); yield* task.halt(); - // Wait for the end file to be written with expected content - let { value: content } = 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}"`); - } - return text; - }, - { timeout: 500 }, - ); - - expect(content).toEqual("goodbye cruel world!"); + let taskError: Error | undefined; + try { + yield* task; + } catch (error) { + taskError = error as Error; + } + expect(taskError?.message).toContain("halted"); }); }); diff --git a/worker/worker.ts b/worker/worker.ts index a3308326..99f59edd 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -217,24 +217,9 @@ export function useWorker( }); 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 }; - if (result.ok) { - resolveOutcome(result.value); - } else { - const serializedError = - result.error as unknown as SerializedError; - rejectOutcome( - errorFromSerialized("Worker failed", serializedError), - ); - } - } - } + worker.terminate(); + rejectOutcome(new Error("worker terminated")); } yield* settled(outcome.operation); }); From a134627b42f5ca8cd7ec4887061d1d4c05f0fc78 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 10:08:52 -0400 Subject: [PATCH 02/12] =?UTF-8?q?=E2=9C=A8=20Make=20worker=20termination?= =?UTF-8?q?=20opt-in?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- worker/README.md | 27 +++++++++++++------- worker/package.json | 2 +- worker/worker.test.ts | 59 +++++++++++++++++++++++++++++++++++++++++-- worker/worker.ts | 20 ++++++++++++--- 4 files changed, 92 insertions(+), 16 deletions(-) diff --git a/worker/README.md b/worker/README.md index f42f8949..83769bfb 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,22 +14,31 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Preemptibly terminates active workers when their host scope shuts down +- 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 that finish on their own deliver their result normally. When the host -scope shuts down while a Worker is still active, `useWorker()` calls -`Worker.terminate()`. Cancellation therefore does not depend on the Worker's -event loop, and can reclaim Workers that are CPU-bound or otherwise -non-cooperative. Worker-side teardown is not guaranteed during cancellation; -durable cleanup should remain owned by the host. +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. -Applications upgrading from `0.5` that relied on Worker-side finalizers during -host cancellation should move that cleanup to the host before upgrading. +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 diff --git a/worker/package.json b/worker/package.json index 3e4bb4a2..bb2d8b8d 100644 --- a/worker/package.json +++ b/worker/package.json @@ -1,6 +1,6 @@ { "name": "@effectionx/worker", - "description": "Web Worker integration with two-way messaging and preemptible cancellation", + "description": "Web Worker integration with two-way messaging and configurable shutdown", "version": "0.6.0", "keywords": ["platform"], "type": "module", diff --git a/worker/worker.test.ts b/worker/worker.test.ts index 25a1383f..b9876661 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -1,5 +1,8 @@ +import { access, mkdir, readFile, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; import { when } from "@effectionx/converge"; -import { describe, it } from "@effectionx/vitest"; +import { beforeEach, describe, it } from "@effectionx/vitest"; import { all, scoped, @@ -11,6 +14,7 @@ import { } from "effection"; import { expect } from "expect"; +import type { ShutdownWorkerParams } from "./test-assets/shutdown-worker.ts"; import { useWorker } from "./worker.ts"; describe("worker", () => { @@ -63,6 +67,57 @@ describe("worker", () => { } }); describe("shutdown", () => { + let startFile: string; + let endFile: string; + let url: string; + + beforeEach(function* () { + let dir = fileURLToPath(import.meta.resolve("./test-tmp")); + yield* until( + rm(dir, { recursive: true, force: true }).then(() => + mkdir(dir, { recursive: true }), + ), + ); + startFile = join(dir, "started.txt"); + endFile = join(dir, "ended.txt"); + url = import.meta.resolve("./test-assets/shutdown-worker.ts"); + }); + + it("shuts down gracefully by default", function* () { + let task = yield* spawn(function* () { + yield* useWorker(url, { + type: "module", + data: { + startFile, + endFile, + endText: "goodbye cruel world!", + } satisfies ShutdownWorkerParams, + }); + yield* suspend(); + }); + + yield* when( + function* () { + let exists = yield* until( + access(startFile).then( + () => true, + () => false, + ), + ); + if (!exists) { + throw new Error("worker has not started"); + } + }, + { timeout: 10_000 }, + ); + + yield* task.halt(); + + 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( @@ -71,7 +126,7 @@ describe("worker", () => { let task = yield* spawn(function* () { yield* useWorker( import.meta.resolve("./test-assets/cpu-bound-worker.ts"), - { type: "module", data: state.buffer }, + { type: "module", data: state.buffer, shutdown: "terminate" }, ); yield* suspend(); }); diff --git a/worker/worker.ts b/worker/worker.ts index 99f59edd..f5ed2c7b 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -69,6 +69,14 @@ export interface WorkerResource ): Operation; } +/** Options for creating and shutting down a Worker. */ +export interface UseWorkerOptions 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 * ``` * * @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 */ export function useWorker( url: string | URL, - options?: WorkerOptions & { data?: TData }, + options?: UseWorkerOptions, ): Operation> { return resource(function* (provide) { let outcome = withResolvers(); @@ -218,8 +226,12 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { - worker.terminate(); - rejectOutcome(new Error("worker terminated")); + if (options?.shutdown === "terminate") { + worker.terminate(); + rejectOutcome(new Error("worker terminated")); + } else { + worker.postMessage({ type: "close" }); + } } yield* settled(outcome.operation); }); From 76c17a99a4dda278a987a472b3453ae99051e954 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:11:09 -0400 Subject: [PATCH 03/12] =?UTF-8?q?=E2=9C=A8=20Add=20contextual=20Worker=20s?= =?UTF-8?q?hutdown=20policies?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pnpm-lock.yaml | 6 +++ worker/README.md | 43 +++++++++++++++---- worker/package.json | 4 +- worker/tsconfig.json | 6 +++ worker/worker.test.ts | 57 ++++++++++++++++++++++++- worker/worker.ts | 97 +++++++++++++++++++++++++++++++++++++++---- 6 files changed, 194 insertions(+), 19 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf415dd1..7edc6aab 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -435,6 +435,12 @@ importers: worker: dependencies: + '@effectionx/context-api': + specifier: workspace:* + version: link:../context-api + '@effectionx/scope-eval': + specifier: workspace:* + version: link:../scope-eval '@effectionx/signals': specifier: workspace:* version: link:../signals diff --git a/worker/README.md b/worker/README.md index 83769bfb..0f123026 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,7 +14,7 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Gracefully shuts down Workers by default with optional forceful termination +- 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 @@ -25,20 +25,47 @@ 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: +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: "terminate", + *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); + }, }); ``` -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. +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 diff --git a/worker/package.json b/worker/package.json index bb2d8b8d..f04ae21a 100644 --- a/worker/package.json +++ b/worker/package.json @@ -1,6 +1,6 @@ { "name": "@effectionx/worker", - "description": "Web Worker integration with two-way messaging and configurable shutdown", + "description": "Web Worker integration with two-way messaging and policy-driven shutdown", "version": "0.6.0", "keywords": ["platform"], "type": "module", @@ -29,6 +29,8 @@ }, "sideEffects": false, "dependencies": { + "@effectionx/context-api": "workspace:*", + "@effectionx/scope-eval": "workspace:*", "@effectionx/signals": "workspace:*", "@effectionx/timebox": "workspace:*", "web-worker": "^1" diff --git a/worker/tsconfig.json b/worker/tsconfig.json index 238fc36a..e2c7a40b 100644 --- a/worker/tsconfig.json +++ b/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" }, diff --git a/worker/worker.test.ts b/worker/worker.test.ts index b9876661..cb7a480b 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -5,6 +5,7 @@ import { when } from "@effectionx/converge"; import { beforeEach, describe, it } from "@effectionx/vitest"; import { all, + createContext, scoped, sleep, spawn, @@ -118,6 +119,54 @@ describe("worker", () => { ); }); + it("cancels its shutdown policy when graceful shutdown completes", function* () { + let shutdownContext = createContext("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 exists = yield* until( + access(startFile).then( + () => true, + () => false, + ), + ); + if (!exists) { + throw new Error("worker has not started"); + } + }, + { 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( @@ -126,7 +175,13 @@ describe("worker", () => { let task = yield* spawn(function* () { yield* useWorker( import.meta.resolve("./test-assets/cpu-bound-worker.ts"), - { type: "module", data: state.buffer, shutdown: "terminate" }, + { + type: "module", + data: state.buffer, + *shutdown(args, terminate) { + return yield* terminate(...args); + }, + }, ); yield* suspend(); }); diff --git a/worker/worker.ts b/worker/worker.ts index f5ed2c7b..859fd1bc 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -1,3 +1,9 @@ +import { + type Api, + type PropertyMiddleware, + createApi, +} from "@effectionx/context-api"; +import { unbox, useEvalScope } from "@effectionx/scope-eval"; import { Err, Ok, @@ -7,6 +13,7 @@ import { ensure, on, once, + race, resource, spawn, withResolvers, @@ -30,6 +37,12 @@ import { */ export interface WorkerResource extends Operation { + /** + * Install middleware that controls whether graceful shutdown escalates to + * hard termination. Calling `next()` terminates the Worker if it is still + * active. + */ + around: Api["around"]; /** * Send a message to the worker and wait for a response. */ @@ -69,12 +82,27 @@ export interface WorkerResource ): Operation; } +/** Context API invoked when an active Worker begins shutting down. */ +export interface WorkerShutdownApi { + /** Hard-terminate the Worker if shutdown middleware delegates to `next()`. */ + shutdown(): Operation; +} + +/** Middleware that controls escalation from graceful shutdown to termination. */ +export type WorkerShutdownMiddleware = PropertyMiddleware< + WorkerShutdownApi, + "shutdown" +>; + /** Options for creating and shutting down a Worker. */ export interface UseWorkerOptions 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"; + /** + * Middleware that may escalate graceful shutdown by calling `next()`, which + * hard-terminates the Worker. Without middleware, shutdown remains graceful. + */ + shutdown?: WorkerShutdownMiddleware; } /** @@ -130,6 +158,11 @@ export function useWorker( options?: UseWorkerOptions, ): Operation> { return resource(function* (provide) { + let { + data, + shutdown: initialShutdownMiddleware, + ...workerOptions + } = options ?? {}; let outcome = withResolvers(); let outcomeSettled = false; @@ -149,7 +182,31 @@ export function useWorker( outcome.reject(error); }; - let worker = new Worker(url, options); + let worker = new Worker(url, workerOptions); + let shutdownApi = createWorkerShutdownApi(() => { + if (!outcomeSettled) { + worker.terminate(); + rejectOutcome(new Error("worker terminated")); + } + }); + let shutdownScope = yield* useEvalScope(); + let hasShutdownMiddleware = false; + + function* around( + ...args: Parameters + ): ReturnType { + let [middlewares] = args; + let result = yield* shutdownScope.eval(() => shutdownApi.around(...args)); + unbox(result); + if (middlewares.shutdown) { + hasShutdownMiddleware = true; + } + } + + if (initialShutdownMiddleware) { + yield* around({ shutdown: initialShutdownMiddleware }); + } + let subscription = yield* on(worker, "message"); // Channel for worker-initiated requests (buffered via eager subscription) @@ -226,11 +283,16 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { - if (options?.shutdown === "terminate") { - worker.terminate(); - rejectOutcome(new Error("worker terminated")); - } else { - worker.postMessage({ type: "close" }); + worker.postMessage({ type: "close" }); + if (hasShutdownMiddleware) { + let result = yield* race([ + settled(outcome.operation), + shutdownScope.eval(() => shutdownApi.operations.shutdown()), + ]); + if (!result.ok && !outcomeSettled) { + worker.terminate(); + rejectOutcome(result.error); + } } } yield* settled(outcome.operation); @@ -238,10 +300,11 @@ export function useWorker( worker.postMessage({ type: "init", - data: options?.data, + data, }); yield* provide({ + around, *send(value) { const response = yield* useChannelResponse(); worker.postMessage( @@ -326,6 +389,22 @@ export function useWorker( }); } +let shutdownApiSequence = 0; + +function createWorkerShutdownApi( + terminate: () => void, +): Api { + let api = createApi( + `@effectionx/worker:shutdown:${shutdownApiSequence++}`, + { + *shutdown(): Operation { + terminate(); + }, + }, + ); + return api; +} + function settled(operation: Operation): Operation> { return { *[Symbol.iterator]() { From 150bca0030463d141c6b9b7f9b9805a950aa04fe Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:45:17 -0400 Subject: [PATCH 04/12] =?UTF-8?q?=F0=9F=90=9B=20Keep=20Worker=20shutdown?= =?UTF-8?q?=20middleware=20scoped?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pnpm-lock.yaml | 3 --- worker/README.md | 4 ++++ worker/package.json | 1 - worker/tsconfig.json | 3 --- worker/worker.test.ts | 51 +++++++++++++++++++++++++++++++++++++++++++ worker/worker.ts | 34 ++++++++++++++++------------- 6 files changed, 74 insertions(+), 22 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7edc6aab..13fac0a0 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -438,9 +438,6 @@ importers: '@effectionx/context-api': specifier: workspace:* version: link:../context-api - '@effectionx/scope-eval': - specifier: workspace:* - version: link:../scope-eval '@effectionx/signals': specifier: workspace:* version: link:../signals diff --git a/worker/README.md b/worker/README.md index 0f123026..ee5010b2 100644 --- a/worker/README.md +++ b/worker/README.md @@ -62,6 +62,10 @@ yield* worker.around({ }); ``` +Middleware installed with `worker.around()` remains active for the Effection +scope that installs it. The `shutdown` construction option remains active for +the Worker's resource lifetime. + 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 diff --git a/worker/package.json b/worker/package.json index f04ae21a..6a11b03c 100644 --- a/worker/package.json +++ b/worker/package.json @@ -30,7 +30,6 @@ "sideEffects": false, "dependencies": { "@effectionx/context-api": "workspace:*", - "@effectionx/scope-eval": "workspace:*", "@effectionx/signals": "workspace:*", "@effectionx/timebox": "workspace:*", "web-worker": "^1" diff --git a/worker/tsconfig.json b/worker/tsconfig.json index e2c7a40b..cd86a1af 100644 --- a/worker/tsconfig.json +++ b/worker/tsconfig.json @@ -13,9 +13,6 @@ { "path": "../converge" }, - { - "path": "../scope-eval" - }, { "path": "../signals" }, diff --git a/worker/worker.test.ts b/worker/worker.test.ts index cb7a480b..039d6248 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -167,6 +167,57 @@ describe("worker", () => { expect(terminated).toEqual(false); }); + it("removes shutdown middleware when its scope exits", function* () { + let terminated = false; + let task = yield* spawn(function* () { + let worker = yield* useWorker(url, { + type: "module", + data: { + startFile, + endFile, + endText: "graceful", + } satisfies ShutdownWorkerParams, + }); + + yield* scoped(function* () { + yield* worker.around({ + *shutdown(args, terminate) { + terminated = true; + return yield* terminate(...args); + }, + }); + }); + + yield* suspend(); + }); + + yield* when( + function* () { + let exists = yield* until( + access(startFile).then( + () => true, + () => false, + ), + ); + if (!exists) { + throw new Error("worker has not started"); + } + }, + { timeout: 10_000 }, + ); + + yield* task.halt(); + + expect(terminated).toEqual(false); + let finalizer = yield* when( + function* () { + return yield* until(readFile(endFile, "utf-8")); + }, + { timeout: 10_000 }, + ); + expect(finalizer.value).toEqual("graceful"); + }); + it("terminates a CPU-bound worker", function* () { expect.assertions(1); let state = new Int32Array( diff --git a/worker/worker.ts b/worker/worker.ts index 859fd1bc..a1133e86 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -3,7 +3,6 @@ import { type PropertyMiddleware, createApi, } from "@effectionx/context-api"; -import { unbox, useEvalScope } from "@effectionx/scope-eval"; import { Err, Ok, @@ -88,6 +87,10 @@ export interface WorkerShutdownApi { shutdown(): Operation; } +interface InternalWorkerShutdownApi extends WorkerShutdownApi { + configured: boolean; +} + /** Middleware that controls escalation from graceful shutdown to termination. */ export type WorkerShutdownMiddleware = PropertyMiddleware< WorkerShutdownApi, @@ -189,18 +192,18 @@ export function useWorker( rejectOutcome(new Error("worker terminated")); } }); - let shutdownScope = yield* useEvalScope(); - let hasShutdownMiddleware = false; function* around( - ...args: Parameters - ): ReturnType { - let [middlewares] = args; - let result = yield* shutdownScope.eval(() => shutdownApi.around(...args)); - unbox(result); - if (middlewares.shutdown) { - hasShutdownMiddleware = true; - } + ...args: Parameters["around"]> + ): ReturnType["around"]> { + let [middlewares, aroundOptions] = args; + yield* shutdownApi.around( + { + shutdown: middlewares.shutdown, + configured: middlewares.shutdown ? () => true : undefined, + }, + aroundOptions, + ); } if (initialShutdownMiddleware) { @@ -284,10 +287,10 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { worker.postMessage({ type: "close" }); - if (hasShutdownMiddleware) { + if (yield* shutdownApi.operations.configured) { let result = yield* race([ settled(outcome.operation), - shutdownScope.eval(() => shutdownApi.operations.shutdown()), + settled(shutdownApi.operations.shutdown()), ]); if (!result.ok && !outcomeSettled) { worker.terminate(); @@ -393,10 +396,11 @@ let shutdownApiSequence = 0; function createWorkerShutdownApi( terminate: () => void, -): Api { - let api = createApi( +): Api { + let api = createApi( `@effectionx/worker:shutdown:${shutdownApiSequence++}`, { + configured: false, *shutdown(): Operation { terminate(); }, From 46ea556302034ee569ba462e51d119422ecaff92 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:52:00 -0400 Subject: [PATCH 05/12] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Configure=20Worker?= =?UTF-8?q?=20shutdown=20at=20creation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- worker/README.md | 14 ++++------- worker/worker.test.ts | 55 +------------------------------------------ worker/worker.ts | 37 +++++------------------------ 3 files changed, 12 insertions(+), 94 deletions(-) diff --git a/worker/README.md b/worker/README.md index ee5010b2..a4d73979 100644 --- a/worker/README.md +++ b/worker/README.md @@ -47,13 +47,13 @@ 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: +Shutdown middleware is selected when the Worker is created but evaluated only +when shutdown begins. It can read context or other application state when +selecting a policy: ```ts -const worker = yield* useWorker("./worker.ts", { type: "module" }); - -yield* worker.around({ +const worker = yield* useWorker("./worker.ts", { + type: "module", *shutdown(args, terminate) { let usage = yield* measureCPUUsage(); yield* sleep(usage < 0.5 ? 10_000 : usage < 0.9 ? 2_000 : 100); @@ -62,10 +62,6 @@ yield* worker.around({ }); ``` -Middleware installed with `worker.around()` remains active for the Effection -scope that installs it. The `shutdown` construction option remains active for -the Worker's resource lifetime. - 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 diff --git a/worker/worker.test.ts b/worker/worker.test.ts index 039d6248..c0355866 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -126,15 +126,13 @@ describe("worker", () => { let policyContext: string | undefined; let terminated = false; let task = yield* spawn(function* () { - let worker = yield* useWorker(url, { + yield* useWorker(url, { type: "module", data: { startFile, endFile, endText: "graceful", } satisfies ShutdownWorkerParams, - }); - yield* worker.around({ *shutdown(_args, terminate) { policyContext = yield* shutdownContext.expect(); yield* suspend(); @@ -167,57 +165,6 @@ describe("worker", () => { expect(terminated).toEqual(false); }); - it("removes shutdown middleware when its scope exits", function* () { - let terminated = false; - let task = yield* spawn(function* () { - let worker = yield* useWorker(url, { - type: "module", - data: { - startFile, - endFile, - endText: "graceful", - } satisfies ShutdownWorkerParams, - }); - - yield* scoped(function* () { - yield* worker.around({ - *shutdown(args, terminate) { - terminated = true; - return yield* terminate(...args); - }, - }); - }); - - yield* suspend(); - }); - - yield* when( - function* () { - let exists = yield* until( - access(startFile).then( - () => true, - () => false, - ), - ); - if (!exists) { - throw new Error("worker has not started"); - } - }, - { timeout: 10_000 }, - ); - - yield* task.halt(); - - expect(terminated).toEqual(false); - let finalizer = yield* when( - function* () { - return yield* until(readFile(endFile, "utf-8")); - }, - { timeout: 10_000 }, - ); - expect(finalizer.value).toEqual("graceful"); - }); - it("terminates a CPU-bound worker", function* () { expect.assertions(1); let state = new Int32Array( diff --git a/worker/worker.ts b/worker/worker.ts index a1133e86..8732496c 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -36,12 +36,6 @@ import { */ export interface WorkerResource extends Operation { - /** - * Install middleware that controls whether graceful shutdown escalates to - * hard termination. Calling `next()` terminates the Worker if it is still - * active. - */ - around: Api["around"]; /** * Send a message to the worker and wait for a response. */ @@ -87,10 +81,6 @@ export interface WorkerShutdownApi { shutdown(): Operation; } -interface InternalWorkerShutdownApi extends WorkerShutdownApi { - configured: boolean; -} - /** Middleware that controls escalation from graceful shutdown to termination. */ export type WorkerShutdownMiddleware = PropertyMiddleware< WorkerShutdownApi, @@ -163,7 +153,7 @@ export function useWorker( return resource(function* (provide) { let { data, - shutdown: initialShutdownMiddleware, + shutdown: shutdownMiddleware, ...workerOptions } = options ?? {}; let outcome = withResolvers(); @@ -193,21 +183,8 @@ export function useWorker( } }); - function* around( - ...args: Parameters["around"]> - ): ReturnType["around"]> { - let [middlewares, aroundOptions] = args; - yield* shutdownApi.around( - { - shutdown: middlewares.shutdown, - configured: middlewares.shutdown ? () => true : undefined, - }, - aroundOptions, - ); - } - - if (initialShutdownMiddleware) { - yield* around({ shutdown: initialShutdownMiddleware }); + if (shutdownMiddleware) { + yield* shutdownApi.around({ shutdown: shutdownMiddleware }); } let subscription = yield* on(worker, "message"); @@ -287,7 +264,7 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { worker.postMessage({ type: "close" }); - if (yield* shutdownApi.operations.configured) { + if (shutdownMiddleware) { let result = yield* race([ settled(outcome.operation), settled(shutdownApi.operations.shutdown()), @@ -307,7 +284,6 @@ export function useWorker( }); yield* provide({ - around, *send(value) { const response = yield* useChannelResponse(); worker.postMessage( @@ -396,11 +372,10 @@ let shutdownApiSequence = 0; function createWorkerShutdownApi( terminate: () => void, -): Api { - let api = createApi( +): Api { + let api = createApi( `@effectionx/worker:shutdown:${shutdownApiSequence++}`, { - configured: false, *shutdown(): Operation { terminate(); }, From d6c93b0eb7797bc56e3f61c5db8b3b5947bd8aea Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 05:02:44 -0400 Subject: [PATCH 06/12] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Make=20Worker=20shut?= =?UTF-8?q?down=20modes=20explicit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pnpm-lock.yaml | 3 -- worker/README.md | 54 +++++++++++----------- worker/package.json | 1 - worker/tsconfig.json | 3 -- worker/worker.test.ts | 101 ++++++++++++++++++++++++++++-------------- worker/worker.ts | 89 ++++++++++++++----------------------- 6 files changed, 129 insertions(+), 122 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 13fac0a0..bf415dd1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -435,9 +435,6 @@ importers: worker: dependencies: - '@effectionx/context-api': - specifier: workspace:* - version: link:../context-api '@effectionx/signals': specifier: workspace:* version: link:../signals diff --git a/worker/README.md b/worker/README.md index a4d73979..d0382b74 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,7 +14,7 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Gracefully shuts down Workers with contextual termination policies +- 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 @@ -26,46 +26,48 @@ Workers shut down gracefully by default. When their host scope shuts down, final result. CPU-bound or otherwise non-cooperative Workers cannot process that close -message. Install shutdown middleware when cancellation must eventually become -preemptible: +message. Use `shutdown: "forced"` when cancellation should terminate the Worker +immediately: ```ts -import { sleep } from "effection"; - const worker = yield* useWorker("./worker.ts", { type: "module", - *shutdown(args, terminate) { - yield* sleep(2_000); - return yield* terminate(...args); - }, + shutdown: "forced", }); ``` -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 selected when the Worker is created but evaluated only -when shutdown begins. It can read context or other application state when -selecting a policy: +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; +} + +const workerHealth = createContext("worker health"); + const worker = yield* useWorker("./worker.ts", { type: "module", - *shutdown(args, terminate) { - let usage = yield* measureCPUUsage(); - yield* sleep(usage < 0.5 ? 10_000 : usage < 0.9 ? 2_000 : 100); - return yield* terminate(...args); + *shutdown() { + const health = yield* workerHealth.expect(); + yield* health.controlChannelUnresponsive; + return "forced"; }, }); ``` -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. +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 diff --git a/worker/package.json b/worker/package.json index 6a11b03c..6196265b 100644 --- a/worker/package.json +++ b/worker/package.json @@ -29,7 +29,6 @@ }, "sideEffects": false, "dependencies": { - "@effectionx/context-api": "workspace:*", "@effectionx/signals": "workspace:*", "@effectionx/timebox": "workspace:*", "web-worker": "^1" diff --git a/worker/tsconfig.json b/worker/tsconfig.json index cd86a1af..238fc36a 100644 --- a/worker/tsconfig.json +++ b/worker/tsconfig.json @@ -7,9 +7,6 @@ "include": ["**/*.ts"], "exclude": ["**/*.test.ts", "test-assets/**", "dist"], "references": [ - { - "path": "../context-api" - }, { "path": "../converge" }, diff --git a/worker/worker.test.ts b/worker/worker.test.ts index c0355866..23e604a5 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -6,6 +6,7 @@ import { beforeEach, describe, it } from "@effectionx/vitest"; import { all, createContext, + type Operation, scoped, sleep, spawn, @@ -16,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* () { @@ -84,6 +89,42 @@ describe("worker", () => { url = import.meta.resolve("./test-assets/shutdown-worker.ts"); }); + function* haltCPUWorker( + shutdown: UseWorkerOptions["shutdown"], + ): Operation { + 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, { @@ -133,11 +174,11 @@ describe("worker", () => { endFile, endText: "graceful", } satisfies ShutdownWorkerParams, - *shutdown(_args, terminate) { + *shutdown() { policyContext = yield* shutdownContext.expect(); yield* suspend(); terminated = true; - return yield* terminate(); + return "forced"; }, }); yield* suspend(); @@ -165,42 +206,34 @@ describe("worker", () => { expect(terminated).toEqual(false); }); - it("terminates a CPU-bound worker", function* () { + it("terminates a CPU-bound worker in forced mode", 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(); + 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; + }>("worker health"); + const controlChannelUnresponsive = withResolvers(); + yield* workerHealth.set({ + controlChannelUnresponsive: controlChannelUnresponsive.operation, }); + controlChannelUnresponsive.resolve(); - yield* when( - function* () { - if (Atomics.load(state, 0) !== 1) { - throw new Error("worker has not started spinning"); - } - }, - { timeout: 10_000 }, - ); + let observedHealth = false; + const shutdown: WorkerShutdownPolicy = function* () { + const health = yield* workerHealth.expect(); + observedHealth = true; + yield* health.controlChannelUnresponsive; + return "forced"; + }; - yield* task.halt(); + const taskError = yield* haltCPUWorker(shutdown); - let taskError: Error | undefined; - try { - yield* task; - } catch (error) { - taskError = error as Error; - } + expect(observedHealth).toEqual(true); expect(taskError?.message).toContain("halted"); }); }); diff --git a/worker/worker.ts b/worker/worker.ts index 8732496c..9d62e4f2 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -1,8 +1,3 @@ -import { - type Api, - type PropertyMiddleware, - createApi, -} from "@effectionx/context-api"; import { Err, Ok, @@ -75,27 +70,18 @@ export interface WorkerResource ): Operation; } -/** Context API invoked when an active Worker begins shutting down. */ -export interface WorkerShutdownApi { - /** Hard-terminate the Worker if shutdown middleware delegates to `next()`. */ - shutdown(): Operation; -} +/** How an active resource should shut down with its owning scope. */ +export type ShutdownMode = "graceful" | "forced"; -/** Middleware that controls escalation from graceful shutdown to termination. */ -export type WorkerShutdownMiddleware = PropertyMiddleware< - WorkerShutdownApi, - "shutdown" ->; +/** Selects a Worker shutdown mode from application state. */ +export type WorkerShutdownPolicy = () => Operation; /** Options for creating and shutting down a Worker. */ export interface UseWorkerOptions extends WorkerOptions { /** Data passed to `workerMain()` during initialization. */ data?: TData; - /** - * Middleware that may escalate graceful shutdown by calling `next()`, which - * hard-terminates the Worker. Without middleware, shutdown remains graceful. - */ - shutdown?: WorkerShutdownMiddleware; + /** Selects graceful, forced, or policy-driven shutdown. */ + shutdown?: ShutdownMode | WorkerShutdownPolicy; } /** @@ -151,11 +137,7 @@ export function useWorker( options?: UseWorkerOptions, ): Operation> { return resource(function* (provide) { - let { - data, - shutdown: shutdownMiddleware, - ...workerOptions - } = options ?? {}; + let { data, shutdown = "graceful", ...workerOptions } = options ?? {}; let outcome = withResolvers(); let outcomeSettled = false; @@ -176,16 +158,12 @@ export function useWorker( }; let worker = new Worker(url, workerOptions); - let shutdownApi = createWorkerShutdownApi(() => { + const terminate = (error = new Error("worker terminated")) => { if (!outcomeSettled) { worker.terminate(); - rejectOutcome(new Error("worker terminated")); + rejectOutcome(error); } - }); - - if (shutdownMiddleware) { - yield* shutdownApi.around({ shutdown: shutdownMiddleware }); - } + }; let subscription = yield* on(worker, "message"); @@ -263,15 +241,25 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { - worker.postMessage({ type: "close" }); - if (shutdownMiddleware) { - let result = yield* race([ - settled(outcome.operation), - settled(shutdownApi.operations.shutdown()), - ]); - if (!result.ok && !outcomeSettled) { - worker.terminate(); - rejectOutcome(result.error); + 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(); + } } } } @@ -368,20 +356,11 @@ export function useWorker( }); } -let shutdownApiSequence = 0; - -function createWorkerShutdownApi( - terminate: () => void, -): Api { - let api = createApi( - `@effectionx/worker:shutdown:${shutdownApiSequence++}`, - { - *shutdown(): Operation { - terminate(); - }, - }, - ); - return api; +function* gracefulCompletion( + outcome: Operation, +): Operation { + yield* outcome; + return "graceful"; } function settled(operation: Operation): Operation> { From 92d8645b8aeaecf482eb5646901159ceb32e0e50 Mon Sep 17 00:00:00 2001 From: Taras Mankovski Date: Tue, 11 Aug 2026 18:05:31 -0400 Subject: [PATCH 07/12] =?UTF-8?q?=E2=9C=A8=20Extract=20@effectionx/forceab?= =?UTF-8?q?le=20and=20implement=20it=20on=20Worker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Graceful teardown assumes the other side is listening. A Worker spinning in a tight loop never reads its control channel, so waiting on it holds the enclosing scope open forever, and no amount of patience fixes it. withForce() puts a deadline on that wait without changing how a resource tears itself down. The resource still runs its own graceful teardown; the policy runs alongside it and may cut it short. Whichever finishes first wins, and a resource that closes in time cancels the policy where it stands. Resources opt in by implementing a single symbol, so one policy shape works across Workers, processes, and anything else holding a handle the runtime will not reclaim on its own. This replaces the shutdown option, its "graceful" | "forced" union, and the policy plumbing inside useWorker, which shrinks by 54 lines. Forcing is quiet: the policy decided to force, so the policy is the natural place to log or count it. Raising from inside a policy is not a reliable alternative — whether the error escapes races against how many turns the resource's teardown needs after being forced. --- forceable/README.md | 98 ++++++++++++++++++++++++++++ forceable/forceable.test.ts | 126 ++++++++++++++++++++++++++++++++++++ forceable/forceable.ts | 113 ++++++++++++++++++++++++++++++++ forceable/mod.ts | 1 + forceable/package.json | 34 ++++++++++ forceable/tsconfig.json | 14 ++++ pnpm-lock.yaml | 12 ++++ pnpm-workspace.yaml | 1 + tsconfig.json | 1 + worker/README.md | 70 ++++++++++---------- worker/package.json | 3 +- worker/tsconfig.json | 3 + worker/worker-force.test.ts | 38 +++++++++++ worker/worker.test.ts | 125 ++++++++++++++++++++++++++--------- worker/worker.ts | 56 +++++----------- 15 files changed, 587 insertions(+), 108 deletions(-) create mode 100644 forceable/README.md create mode 100644 forceable/forceable.test.ts create mode 100644 forceable/forceable.ts create mode 100644 forceable/mod.ts create mode 100644 forceable/package.json create mode 100644 forceable/tsconfig.json create mode 100644 worker/worker-force.test.ts diff --git a/forceable/README.md b/forceable/README.md new file mode 100644 index 00000000..86126c4b --- /dev/null +++ b/forceable/README.md @@ -0,0 +1,98 @@ +# Forceable + +Put a deadline on a resource's graceful teardown. + +--- + +Well behaved resources shut down cooperatively: they ask, then wait. A worker is +told to close and is given time to finish; a process is sent a signal and is +given time to flush. Waiting is the right default, because it is the only way +the resource gets to run its own cleanup. + +But cooperative shutdown assumes the other side is listening. A worker spinning +in a tight loop never reads its control channel. A process ignoring `SIGTERM` +never flushes. Waiting on either one holds the enclosing scope open forever, and +no amount of patience fixes it. + +`withForce` puts a deadline on that wait, without changing how the resource +tears itself down. + +```ts +import { sleep } from "effection"; +import { withForce } from "@effectionx/forceable"; +import { useWorker } from "@effectionx/worker"; + +let worker = yield* withForce( + useWorker("./transcode.ts", { type: "module" }), + function* (force) { + yield* sleep(10_000); + force("worker did not close within 10s"); + }, +); +``` + +The resource still runs its own graceful teardown, exactly as it would have. The +policy runs alongside it. Whichever finishes first wins: if the resource closes +in time, the policy is cancelled wherever it happens to be suspended and never +forces; if the policy calls `force` first, the graceful teardown is cut short. + +## Policies read application state + +A policy is an operation, so it can wait on anything — not just a clock. That +matters because a fixed timeout is a guess, while the application usually knows +something more specific about whether shutdown is going to land. + +```ts +let worker = yield* withForce(useWorker(url, { type: "module" }), function* (force) { + let health = yield* WorkerHealth.expect(); + yield* health.controlChannelUnresponsive; + force("control channel stopped answering"); +}); +``` + +This package deliberately does not infer that CPU load, message latency, or +elapsed time mean a particular resource is unhealthy. Applications define that. + +## Implementing Forceable + +A resource opts in by implementing one symbol. It should tear down immediately, +tolerate being called more than once, and do nothing if the resource has already +finished. + +```ts +import { type Forceable, ForcedTerminationError, force } from "@effectionx/forceable"; + +yield* provide({ + [force](reason?: string) { + handle.destroy(); + rejectOutcome(new ForcedTerminationError(reason)); + }, +}); +``` + +Anything implementing `Forceable` composes with `withForce`, so a single policy +shape works across workers, processes, and anything else holding a handle the +runtime will not reclaim on its own. + +## What forcing costs + +Forcing skips the resource's cleanup. That is the entire point, and it is not +free: whatever the graceful teardown was responsible for — flushing a buffer, +removing a lock file, acknowledging a message — has not happened. Durable +cleanup for a resource that might be forced has to be owned by the host. + +Forcing is also quiet. `force(reason)` returns, teardown finishes, and the +enclosing halt is undisturbed. The policy is the notification: it decided to +force, so it is the natural place to log, count, or alert. + +```ts +function* (force) { + yield* sleep(grace); + logger.warn({ reason }, "forced teardown"); + force(reason); +} +``` + +Raising from inside a policy is **not** a reliable way to make forcing loud. +Whether the error escapes is a race against how many turns the resource's own +teardown needs after being forced. diff --git a/forceable/forceable.test.ts b/forceable/forceable.test.ts new file mode 100644 index 00000000..ddd161dd --- /dev/null +++ b/forceable/forceable.test.ts @@ -0,0 +1,126 @@ +import { describe, it } from "@effectionx/vitest"; +import { + type Operation, + ensure, + resource, + scoped, + sleep, + spawn, + suspend, + withResolvers, +} from "effection"; +import { expect } from "expect"; + +import { + ForcedTerminationError, + type Forceable, + force, + withForce, +} from "./forceable.ts"; + +interface Stubborn extends Forceable { + /** how the teardown ended, readable after the scope is gone */ + report(): string; +} + +/** + * A resource whose graceful teardown takes `closeAfter` milliseconds and which + * can be cut short. Stands in for anything holding an operating system handle. + */ +function useStubborn(closeAfter: number, log: string[]): Operation { + return resource(function* (provide) { + let settled = withResolvers(); + let done = false; + + const finish = (how: string) => { + if (!done) { + done = true; + log.push(how); + settled.resolve(); + } + }; + + yield* ensure(function* () { + yield* spawn(function* () { + yield* sleep(closeAfter); + finish("closed gracefully"); + }); + yield* settled.operation; + }); + + yield* provide({ + [force]: (reason?: string) => finish(`forced: ${reason}`), + report: () => log.join(), + }); + }); +} + +describe("withForce", () => { + it("lets graceful teardown finish when it lands inside the deadline", function* () { + let log: string[] = []; + + yield* scoped(function* () { + yield* withForce(useStubborn(10, log), function* (force) { + yield* sleep(200); + force("deadline expired"); + }); + }); + + expect(log).toEqual(["closed gracefully"]); + }); + + it("cuts graceful teardown short when the deadline expires first", function* () { + let log: string[] = []; + + yield* scoped(function* () { + yield* withForce(useStubborn(10_000, log), function* (force) { + yield* sleep(10); + force("deadline expired"); + }); + }); + + expect(log).toEqual(["forced: deadline expired"]); + }); + + it("cancels the policy so it cannot force after a graceful teardown", function* () { + let log: string[] = []; + let policyFinished = false; + + yield* scoped(function* () { + yield* withForce(useStubborn(10, log), function* (force) { + yield* suspend(); + policyFinished = true; + force("should never happen"); + }); + }); + + expect(log).toEqual(["closed gracefully"]); + expect(policyFinished).toEqual(false); + }); + + it("stays quiet, so forcing does not disturb the halt", function* () { + let log: string[] = []; + let halted: Error | undefined; + + let task = yield* spawn(function* () { + yield* withForce(useStubborn(10_000, log), function* (force) { + force("deadline expired"); + }); + yield* suspend(); + }); + + yield* sleep(50); + try { + yield* task.halt(); + } catch (error) { + halted = error as Error; + } + + expect(halted).toBeUndefined(); + }); + + // Raising from inside a policy is NOT a reliable way to make forcing loud: + // whether the throw lands is a race against how many turns the resource's own + // teardown needs after being forced. If forcing should be loud, that belongs + // in withForce, where it can be deterministic. +}); diff --git a/forceable/forceable.ts b/forceable/forceable.ts new file mode 100644 index 00000000..46c177f8 --- /dev/null +++ b/forceable/forceable.ts @@ -0,0 +1,113 @@ +import { + type Operation, + ensure, + resource, + scoped, + spawn, + suspend, + withResolvers, +} from "effection"; + +/** + * Abandon a resource's graceful teardown and tear it down immediately. + * + * A resource implements this when it can be shut down two ways: cooperatively, + * by asking and waiting, and forcibly, by seizing whatever the operating system + * gave it. Cooperative shutdown is always attempted first and is often the only + * one that runs. + */ +export const force = Symbol.for("effection.force"); + +/** + * Reported to whoever is waiting on a resource that was torn down forcibly. + * Its graceful teardown never ran, so anything that teardown was responsible + * for is still outstanding. + */ +export class ForcedTerminationError extends Error { + override name = "ForcedTerminationError"; + + constructor(reason?: string) { + super(reason ?? "torn down forcibly"); + } +} + +/** A resource whose graceful teardown can be abandoned. */ +export interface Forceable { + /** + * Tear down immediately. Safe to call more than once and safe to call on a + * resource that already finished — later calls do nothing. + * + * @param reason why the graceful teardown was abandoned, for the error the + * resource reports to whoever is waiting on it + */ + [force](reason?: string): void; +} + +/** + * A policy decides when graceful teardown has gone on long enough. It begins + * when teardown begins and runs alongside it, so it can read application state + * as the shutdown actually unfolds. Calling `force` abandons the graceful + * teardown; returning without calling it waits however long that takes. + * + * If the resource tears down gracefully first, the policy is cancelled wherever + * it happens to be suspended. + */ +export type ForcePolicy = (force: (reason?: string) => void) => Operation; + +/** + * Give a resource a deadline, so a graceful teardown that never lands cannot + * hold its scope open forever. + * + * The resource keeps its own graceful teardown and runs it exactly as it would + * have. `policy` runs concurrently with that teardown and may cut it short. + * + * @example Give a stubborn resource ten seconds, then take it down + * ```ts + * import { sleep } from "effection"; + * import { withForce } from "@effectionx/forceable"; + * + * let connection = yield* withForce(useConnection(url), function* (force) { + * yield* sleep(10_000); + * force("connection did not drain within 10s"); + * }); + * ``` + * + * @example Decide from application state rather than a clock + * ```ts + * let connection = yield* withForce(useConnection(url), function* (force) { + * let health = yield* Health.expect(); + * yield* health.unresponsive; + * force("stopped answering health checks"); + * }); + * ``` + * + * @param op operation acquiring the resource to put a deadline on + * @param policy decides when to abandon the graceful teardown + */ +export function withForce( + op: Operation, + policy: ForcePolicy, +): Operation { + return resource(function* (provide) { + let acquired = withResolvers(); + + // Hold the resource open in a child task so that halting the task, rather + // than exiting this frame, is what starts its graceful teardown. That gives + // the policy below something to run alongside. + let held = yield* spawn(function* () { + acquired.resolve(yield* op); + yield* suspend(); + }); + + let value = yield* acquired.operation; + + yield* ensure(function* () { + yield* scoped(function* () { + yield* spawn(() => policy((reason) => value[force](reason))); + yield* held.halt(); + }); + }); + + yield* provide(value); + }); +} diff --git a/forceable/mod.ts b/forceable/mod.ts new file mode 100644 index 00000000..171e1934 --- /dev/null +++ b/forceable/mod.ts @@ -0,0 +1 @@ +export * from "./forceable.ts"; diff --git a/forceable/package.json b/forceable/package.json new file mode 100644 index 00000000..cb2afc61 --- /dev/null +++ b/forceable/package.json @@ -0,0 +1,34 @@ +{ + "name": "@effectionx/forceable", + "description": "Put a deadline on a resource's graceful teardown and tear it down forcibly when it expires", + "version": "0.1.0", + "keywords": ["concurrency"], + "type": "module", + "main": "./dist/mod.js", + "types": "./dist/mod.d.ts", + "exports": { + ".": { + "types": "./dist/mod.d.ts", + "development": "./mod.ts", + "import": "./dist/mod.js", + "default": "./dist/mod.js" + } + }, + "peerDependencies": { + "effection": "^3 || ^4" + }, + "license": "MIT", + "author": "engineering@frontside.com", + "repository": { + "type": "git", + "url": "git+https://github.com/thefrontside/effectionx.git" + }, + "bugs": { + "url": "https://github.com/thefrontside/effectionx/issues" + }, + "sideEffects": false, + "devDependencies": { + "@effectionx/vitest": "workspace:*", + "effection": "^4" + } +} diff --git a/forceable/tsconfig.json b/forceable/tsconfig.json new file mode 100644 index 00000000..49b10377 --- /dev/null +++ b/forceable/tsconfig.json @@ -0,0 +1,14 @@ +{ + "extends": "../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "." + }, + "include": ["**/*.ts"], + "exclude": ["**/*.test.ts", "dist"], + "references": [ + { + "path": "../vitest" + } + ] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf415dd1..619855f3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -136,6 +136,15 @@ importers: specifier: ^4 version: 4.1.0 + forceable: + devDependencies: + '@effectionx/vitest': + specifier: workspace:* + version: link:../vitest + effection: + specifier: ^4 + version: 4.1.0 + fs: dependencies: '@effectionx/context-api': @@ -435,6 +444,9 @@ importers: worker: dependencies: + '@effectionx/forceable': + specifier: workspace:* + version: link:../forceable '@effectionx/signals': specifier: workspace:* version: link:../signals diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 2a225379..fffbf009 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -7,6 +7,7 @@ packages: # deno-deploy excluded - deprecated Deno-only package - "effect-ts" - "fetch" + - "forceable" - "fs" - "fx" - "inline" diff --git a/tsconfig.json b/tsconfig.json index 4c1ab0b0..60a47cb0 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -19,6 +19,7 @@ { "path": "converge" }, { "path": "effect-ts" }, { "path": "fetch" }, + { "path": "forceable" }, { "path": "fs" }, { "path": "fx" }, { "path": "inline" }, diff --git a/worker/README.md b/worker/README.md index d0382b74..576c842f 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,60 +14,60 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Gracefully or forcibly shuts down Workers with contextual policies +- Shuts down gracefully, with an opt-in deadline for Workers that will not close - 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. +Workers shut down gracefully. When the host scope shuts down, `useWorker()` +sends a close message and waits for Worker-side teardown and the final result, +so the Worker gets to run its own cleanup. -CPU-bound or otherwise non-cooperative Workers cannot process that close -message. Use `shutdown: "forced"` when cancellation should terminate the Worker -immediately: +That waiting assumes the Worker is listening. A Worker busy in a tight loop +never reads the close message, and waiting on it holds the host scope open +forever. -```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: +`useWorker()` implements [`@effectionx/forceable`][forceable], so wrap it in +`withForce()` to put a deadline on that wait: ```ts -import { createContext, type Operation } from "effection"; +import { sleep } from "effection"; +import { withForce } from "@effectionx/forceable"; +import { useWorker } from "@effectionx/worker"; -interface WorkerHealth { - controlChannelUnresponsive: Operation; -} +const worker = yield* withForce( + useWorker("./worker.ts", { type: "module" }), + function* (force) { + yield* sleep(10_000); + force("worker did not close within 10s"); + }, +); +``` -const workerHealth = createContext("worker health"); +A policy is an operation, so it can wait on application state instead of a +clock. This package does not infer that CPU use or delayed messages mean a +particular Worker is unhealthy — applications define that: -const worker = yield* useWorker("./worker.ts", { - type: "module", - *shutdown() { +```ts +const worker = yield* withForce( + useWorker("./worker.ts", { type: "module" }), + function* (force) { const health = yield* workerHealth.expect(); yield* health.controlChannelUnresponsive; - return "forced"; + force("control channel stopped answering"); }, -}); +); ``` -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. +If the Worker closes before the policy forces, the policy is cancelled where it +stands and teardown finishes normally. No timeout is imposed by default. + +Forcing calls `Worker.terminate()`, which does not run Worker-side finalizers, +so durable cleanup for a Worker that might be forced must be owned by the host. -Forced termination does not run Worker-side finalizers, so durable cleanup for -a forcibly terminated Worker must be owned by the host. +[forceable]: ../forceable/README.md ## Usage: Get worker's return value diff --git a/worker/package.json b/worker/package.json index 6196265b..b5c4c26c 100644 --- a/worker/package.json +++ b/worker/package.json @@ -1,6 +1,6 @@ { "name": "@effectionx/worker", - "description": "Web Worker integration with two-way messaging and policy-driven shutdown", + "description": "Web Worker integration with two-way messaging and structured shutdown", "version": "0.6.0", "keywords": ["platform"], "type": "module", @@ -29,6 +29,7 @@ }, "sideEffects": false, "dependencies": { + "@effectionx/forceable": "workspace:*", "@effectionx/signals": "workspace:*", "@effectionx/timebox": "workspace:*", "web-worker": "^1" diff --git a/worker/tsconfig.json b/worker/tsconfig.json index 238fc36a..6923df0f 100644 --- a/worker/tsconfig.json +++ b/worker/tsconfig.json @@ -10,6 +10,9 @@ { "path": "../converge" }, + { + "path": "../forceable" + }, { "path": "../signals" }, diff --git a/worker/worker-force.test.ts b/worker/worker-force.test.ts new file mode 100644 index 00000000..94f8ea0e --- /dev/null +++ b/worker/worker-force.test.ts @@ -0,0 +1,38 @@ +import { describe, it } from "@effectionx/vitest"; +import { scoped, sleep, spawn, suspend } from "effection"; +import { expect } from "expect"; + +import { ForcedTerminationError, withForce } from "@effectionx/forceable"; +import { useWorker } from "./worker.ts"; + +/** A worker that spins forever and never services the close message. */ +function spinning() { + let state = new Int32Array(new SharedArrayBuffer(4)); + return useWorker(import.meta.resolve("./test-assets/cpu-bound-worker.ts"), { + type: "module", + data: state.buffer, + }); +} + +describe("withForce", () => { + it("stays quiet by default, so forcing does not disturb the halt", function* () { + expect.assertions(1); + let halted: Error | undefined; + + let task = yield* spawn(function* () { + yield* withForce(spinning(), function* (force) { + force("cpu bound"); + }); + yield* suspend(); + }); + + yield* sleep(200); + try { + yield* task.halt(); + } catch (error) { + halted = error as Error; + } + + expect(halted).toBeUndefined(); + }); +}); diff --git a/worker/worker.test.ts b/worker/worker.test.ts index 23e604a5..30aecfd9 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -16,12 +16,9 @@ import { } from "effection"; import { expect } from "expect"; +import { type ForcePolicy, withForce } from "@effectionx/forceable"; import type { ShutdownWorkerParams } from "./test-assets/shutdown-worker.ts"; -import { - type UseWorkerOptions, - type WorkerShutdownPolicy, - useWorker, -} from "./worker.ts"; +import { useWorker } from "./worker.ts"; describe("worker", () => { it("sends and receive messages in synchrony", function* () { @@ -72,6 +69,23 @@ describe("worker", () => { expect((e as Error).message).toContain("boom!"); } }); + it("propagates a worker error through withForce", function* () { + expect.assertions(2); + let worker = yield* withForce( + useWorker(import.meta.resolve("./test-assets/boom-result-worker.ts"), { + type: "module", + data: "boom!", + }), + function* () {}, + ); + + try { + yield* worker; + } catch (e) { + expect(e).toBeInstanceOf(Error); + expect((e as Error).message).toContain("boom!"); + } + }); describe("shutdown", () => { let startFile: string; let endFile: string; @@ -89,20 +103,17 @@ describe("worker", () => { url = import.meta.resolve("./test-assets/shutdown-worker.ts"); }); - function* haltCPUWorker( - shutdown: UseWorkerOptions["shutdown"], - ): Operation { + function* haltCPUWorker(policy: ForcePolicy): Operation { 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"), - { + yield* withForce( + useWorker(import.meta.resolve("./test-assets/cpu-bound-worker.ts"), { type: "module", data: state.buffer, - shutdown, - }, + }), + policy, ); yield* suspend(); }); @@ -160,27 +171,29 @@ describe("worker", () => { ); }); - it("cancels its shutdown policy when graceful shutdown completes", function* () { + it("cancels its force policy when graceful shutdown completes", function* () { let shutdownContext = createContext("worker shutdown test"); yield* shutdownContext.set("available during shutdown"); let policyContext: string | undefined; - let terminated = false; + let forced = false; let task = yield* spawn(function* () { - yield* useWorker(url, { - type: "module", - data: { - startFile, - endFile, - endText: "graceful", - } satisfies ShutdownWorkerParams, - *shutdown() { + yield* withForce( + useWorker(url, { + type: "module", + data: { + startFile, + endFile, + endText: "graceful", + } satisfies ShutdownWorkerParams, + }), + function* (force) { policyContext = yield* shutdownContext.expect(); yield* suspend(); - terminated = true; - return "forced"; + forced = true; + force("should never happen"); }, - }); + ); yield* suspend(); }); @@ -203,12 +216,14 @@ describe("worker", () => { expect(yield* until(readFile(endFile, "utf-8"))).toEqual("graceful"); expect(policyContext).toEqual("available during shutdown"); - expect(terminated).toEqual(false); + expect(forced).toEqual(false); }); - it("terminates a CPU-bound worker in forced mode", function* () { + it("forces a CPU-bound worker that cannot service the close message", function* () { expect.assertions(1); - const taskError = yield* haltCPUWorker("forced"); + const taskError = yield* haltCPUWorker(function* (force) { + force("cpu bound"); + }); expect(taskError?.message).toContain("halted"); }); @@ -224,18 +239,64 @@ describe("worker", () => { controlChannelUnresponsive.resolve(); let observedHealth = false; - const shutdown: WorkerShutdownPolicy = function* () { + const policy: ForcePolicy = function* (force) { const health = yield* workerHealth.expect(); observedHealth = true; yield* health.controlChannelUnresponsive; - return "forced"; + force("control channel unresponsive"); }; - const taskError = yield* haltCPUWorker(shutdown); + const taskError = yield* haltCPUWorker(policy); expect(observedHealth).toEqual(true); expect(taskError?.message).toContain("halted"); }); + + // Documents a gap rather than a guarantee. Forcing settles the outcome as a + // ForcedTerminationError, but an awaiter inside the halted scope is cancelled + // before it can observe that rejection, and useWorker's own teardown swallows + // it via settled(). So the reason reaches nobody. Whether it should escape — + // and at the cost of masking an in-flight error — is the open question. + it("does not surface the force reason to an awaiter", function* () { + expect.assertions(1); + let state = new Int32Array( + new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT), + ); + let outcome: Error | undefined; + + let task = yield* spawn(function* () { + let worker = yield* withForce( + useWorker(import.meta.resolve("./test-assets/cpu-bound-worker.ts"), { + type: "module", + data: state.buffer, + }), + function* (force) { + force("event loop p99 300ms"); + }, + ); + yield* spawn(function* () { + try { + yield* worker; + } catch (error) { + outcome = error as Error; + } + }); + 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(); + + expect(outcome).toBeUndefined(); + }); }); it("becomes halted if you try and await its value out of scope", function* () { diff --git a/worker/worker.ts b/worker/worker.ts index 9d62e4f2..37367565 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -7,7 +7,6 @@ import { ensure, on, once, - race, resource, spawn, withResolvers, @@ -15,6 +14,11 @@ import { import Worker from "web-worker"; import { useChannelRequest, useChannelResponse } from "./channel.ts"; +import { + ForcedTerminationError, + type Forceable, + force, +} from "@effectionx/forceable"; import { type ForEachContext, type SerializedError, @@ -30,7 +34,8 @@ import { * @template TReturn - worker operation return value */ export interface WorkerResource - extends Operation { + extends Operation, + Forceable { /** * Send a message to the worker and wait for a response. */ @@ -70,18 +75,10 @@ export interface WorkerResource ): Operation; } -/** 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; - -/** Options for creating and shutting down a Worker. */ +/** Options for creating a Worker. */ export interface UseWorkerOptions extends WorkerOptions { /** Data passed to `workerMain()` during initialization. */ data?: TData; - /** Selects graceful, forced, or policy-driven shutdown. */ - shutdown?: ShutdownMode | WorkerShutdownPolicy; } /** @@ -137,7 +134,7 @@ export function useWorker( options?: UseWorkerOptions, ): Operation> { return resource(function* (provide) { - let { data, shutdown = "graceful", ...workerOptions } = options ?? {}; + let { data, ...workerOptions } = options ?? {}; let outcome = withResolvers(); let outcomeSettled = false; @@ -241,28 +238,10 @@ export function useWorker( yield* ensure(function* () { if (!outcomeSettled) { - 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(); - } - } - } + worker.postMessage({ type: "close" }); } + // A worker that cannot service the close message never settles this. + // Wrap with withForce() to put a deadline on it. yield* settled(outcome.operation); }); @@ -351,18 +330,15 @@ export function useWorker( } }, + [force](reason?: string) { + terminate(new ForcedTerminationError(reason)); + }, + [Symbol.iterator]: outcome.operation[Symbol.iterator], }); }); } -function* gracefulCompletion( - outcome: Operation, -): Operation { - yield* outcome; - return "graceful"; -} - function settled(operation: Operation): Operation> { return { *[Symbol.iterator]() { From bfeef62deeb75fd33af336b0d2a34ff086a28535 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:33:28 -0400 Subject: [PATCH 08/12] =?UTF-8?q?=F0=9F=93=9D=20Record=20why=20teardown=20?= =?UTF-8?q?can=20wait=20on=20the=20message=20loop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Graceful teardown used to read the close message inline, duplicating the spawned message loop that was already handling it. Removing the duplicate leaves teardown depending on that loop still being alive, which is true but invisible. Say so, and say which effection versions it was checked against, so the next person weighing a teardown change knows what is holding it up. --- worker/worker.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/worker/worker.ts b/worker/worker.ts index 37367565..a7facdaf 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -240,8 +240,13 @@ export function useWorker( if (!outcomeSettled) { worker.postMessage({ type: "close" }); } - // A worker that cannot service the close message never settles this. - // Wrap with withForce() to put a deadline on it. + // Settled by the spawned message loop above, which is still running: + // ensure handlers run before a resource's spawned children are halted. + // Verified against effection 3.0.0, 3.6.1, and 4.1.0 — see test:matrix. + // + // A Worker that cannot service the close message never settles this, and + // waiting on it holds this scope open forever. Wrap the resource in + // withForce() from @effectionx/forceable to put a deadline on the wait. yield* settled(outcome.operation); }); From f2ca4304dfc1e1ccb3a2b7410da488ff96a08efa Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:03:14 -0400 Subject: [PATCH 09/12] =?UTF-8?q?=E2=9C=A8=20Implement=20Forceable=20on=20?= =?UTF-8?q?exec()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A process that traps SIGTERM, or a descendant holding the inherited stdio open, never lets teardown finish: exec() signals the group and then waits on stdout and stderr closing, and that wait has no bound. Implement the force symbol so withForce() can put one there. POSIX sends SIGKILL to the process group, which cannot be trapped and reaches descendants holding the stdio open. Windows runs taskkill /T /F from a task owned by the resource, so cancelling whatever policy asked for the kill cannot cancel the kill itself. daemon() returns the same Process, so it gains this too. --- process/package.json | 1 + process/src/exec/posix.ts | 15 +++++ process/src/exec/types.ts | 3 +- process/src/exec/win32.ts | 16 +++++ process/test/fixtures/shutdown-resistant.ts | 13 ++++ process/test/force.test.ts | 73 +++++++++++++++++++++ process/tsconfig.json | 3 + 7 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 process/test/fixtures/shutdown-resistant.ts create mode 100644 process/test/force.test.ts diff --git a/process/package.json b/process/package.json index 6215178a..132760fa 100644 --- a/process/package.json +++ b/process/package.json @@ -30,6 +30,7 @@ "sideEffects": false, "dependencies": { "@effectionx/context-api": "workspace:*", + "@effectionx/forceable": "workspace:*", "@effectionx/node": "workspace:*", "@effectionx/scope-eval": "workspace:*", "cross-spawn": "^7", diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index 6b5bb9bc..1c15f672 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -12,6 +12,7 @@ import { spawn, withResolvers, } from "effection"; +import { ForcedTerminationError, force } from "@effectionx/forceable"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; import { once } from "@effectionx/node/events"; import { fromReadable } from "@effectionx/node/stream"; @@ -129,6 +130,8 @@ export function* createPosixProcess( throw new Error("no pid for childProcess"); } process.kill(-childProcess.pid, "SIGTERM"); + // A process that traps SIGTERM, or a descendant holding the inherited + // stdio open, never settles this. Wrap with withForce() to bound it. yield* all([io.stdoutDone.operation, io.stderrDone.operation]); } catch (_e) { // do nothing, process is probably already dead @@ -137,6 +140,18 @@ export function* createPosixProcess( return { pid: pid as number, + [force](reason?: string) { + // SIGKILL cannot be trapped, and addressing the group reaches + // descendants that are holding the inherited stdio open. + try { + if (typeof childProcess.pid === "number") { + process.kill(-childProcess.pid, "SIGKILL"); + } + } catch (_e) { + // already gone + } + processResult.resolve(Err(new ForcedTerminationError(reason))); + }, *around( ...args: Parameters ): ReturnType { diff --git a/process/src/exec/types.ts b/process/src/exec/types.ts index a45e75d4..1949c6d2 100644 --- a/process/src/exec/types.ts +++ b/process/src/exec/types.ts @@ -1,3 +1,4 @@ +import type { Forceable } from "@effectionx/forceable"; import type { Operation } from "effection"; import type { OutputStream } from "../helpers.ts"; import type { Api } from "@effectionx/context-api"; @@ -13,7 +14,7 @@ export interface Writable { * The process type is what is returned by the `exec` operation. It has all of * standard io handles, and methods for synchronizing on return. */ -export interface Process extends StdIO { +export interface Process extends StdIO, Forceable { /** Child process id as reported by the operating system. */ readonly pid: number; diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts index 3a2c78b2..bc9755e4 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -25,6 +25,7 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; +import { ForcedTerminationError, force } from "@effectionx/forceable"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; type ProcessResultValue = [number?, string?]; @@ -154,6 +155,18 @@ export function* createWin32Process( } }); + // taskkill is an operation, but [force] is synchronous, so the kill runs in + // a task owned by this resource. Cancelling whatever policy called [force] + // therefore cannot cancel the kill it asked for. + let forced = withResolvers(); + yield* spawn(function* () { + let reason = yield* forced.operation; + if (pid) { + yield* killTree(pid); + } + processResult.resolve(Err(new ForcedTerminationError(reason))); + }); + yield* ensure(function* () { // If no pid is available, we have no way to kill the process, // so we skip and presume it is cleaned up. @@ -191,6 +204,9 @@ export function* createWin32Process( return { pid: pid as number, + [force](reason?: string) { + forced.resolve(reason); + }, *around( ...args: Parameters ): ReturnType { diff --git a/process/test/fixtures/shutdown-resistant.ts b/process/test/fixtures/shutdown-resistant.ts new file mode 100644 index 00000000..f05ea05c --- /dev/null +++ b/process/test/fixtures/shutdown-resistant.ts @@ -0,0 +1,13 @@ +import process from "node:process"; + +process.on("SIGINT", () => {}); +process.on("SIGTERM", () => {}); + +if (process.platform !== "win32") { + process.on("SIGUSR1", () => { + process.exit(0); + }); +} + +console.log("ready"); +setInterval(() => {}, 1_000); diff --git a/process/test/force.test.ts b/process/test/force.test.ts new file mode 100644 index 00000000..0f3e20bc --- /dev/null +++ b/process/test/force.test.ts @@ -0,0 +1,73 @@ +import process from "node:process"; +import { withForce } from "@effectionx/forceable"; +import { describe, it } from "@effectionx/vitest"; +import { type Task, sleep, spawn, suspend } from "effection"; +import { expect } from "expect"; + +import { exec } from "../mod.ts"; + +/** + * Traps SIGTERM and never exits, so graceful teardown alone would wait forever. + */ +const resistant = () => `${process.execPath} ./fixtures/shutdown-resistant.ts`; + +describe("withForce(exec())", () => { + it("bounds a process that ignores graceful shutdown", function* () { + let pid: number | undefined; + + let task: Task = yield* spawn(function* () { + let proc = yield* withForce( + exec(resistant(), { cwd: import.meta.dirname }), + function* (force) { + yield* sleep(100); + force("ignored SIGTERM"); + }, + ); + pid = proc.pid; + yield* suspend(); + }); + + // Give the child time to install its signal handlers. + yield* sleep(500); + expect(pid).toBeDefined(); + + // Without withForce this halt never returns: the child traps SIGTERM and + // teardown waits on stdio that never closes. + yield* task.halt(); + + // SIGKILL went to the group, so the process is actually gone rather than + // merely abandoned. kill(pid, 0) throws ESRCH once it has been reaped. + yield* sleep(100); + let alive = true; + try { + process.kill(pid as number, 0); + } catch { + alive = false; + } + expect(alive).toBe(false); + }); + + it("leaves a cooperative process to shut down gracefully", function* () { + let forced = false; + + let task: Task = yield* spawn(function* () { + yield* withForce( + exec(`${process.execPath} -e "setInterval(() => {}, 1000)"`, { + cwd: import.meta.dirname, + }), + function* (force) { + yield* sleep(10_000); + forced = true; + force("should not happen"); + }, + ); + yield* suspend(); + }); + + yield* sleep(300); + yield* task.halt(); + + // SIGTERM was enough, so the policy was cancelled where it stood. + expect(forced).toBe(false); + }); +}); diff --git a/process/tsconfig.json b/process/tsconfig.json index bdfb17ce..286a5b17 100644 --- a/process/tsconfig.json +++ b/process/tsconfig.json @@ -10,6 +10,9 @@ { "path": "../context-api" }, + { + "path": "../forceable" + }, { "path": "../node" }, From 10c3ff236c7f307da525b65eca51a0af83421a35 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:06:56 -0400 Subject: [PATCH 10/12] =?UTF-8?q?=F0=9F=93=9D=20Document=20process=20shutd?= =?UTF-8?q?own=20and=20bound=20it=20in=20the=20README?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Explain what cooperative shutdown waits for, the two ordinary cases that never satisfy that wait, and how withForce() bounds it. Record the force symbol alongside join() and expect() in the Process interface. Minor bump: exec() and daemon() gain a capability without changing any existing behaviour. --- pnpm-lock.yaml | 3 +++ process/README.md | 59 ++++++++++++++++++++++++++++++++++++++++++++ process/package.json | 2 +- 3 files changed, 63 insertions(+), 1 deletion(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 619855f3..b0e12437 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -214,6 +214,9 @@ importers: '@effectionx/context-api': specifier: workspace:* version: link:../context-api + '@effectionx/forceable': + specifier: workspace:* + version: link:../forceable '@effectionx/node': specifier: workspace:* version: link:../node diff --git a/process/README.md b/process/README.md index 45cc1c55..72b7e21e 100644 --- a/process/README.md +++ b/process/README.md @@ -13,6 +13,7 @@ finite lifetime, and `daemon()` for long-running processes like servers. - Stream-based access to stdout and stderr - Writable stdin for sending input to processes - Proper signal handling and cleanup on both POSIX and Windows +- Optional deadline on shutdown for processes that will not exit - Shell mode for complex commands with glob expansion - Structured error handling with `join()` and `expect()` methods @@ -153,6 +154,60 @@ await main(function* () { }); ``` +## Shutdown + +When the owning scope exits, a process is shut down cooperatively: `SIGTERM` to +the process group on POSIX, Ctrl-C plus stdin closure on Windows. Teardown then +waits for the process to exit and for its captured stdout and stderr to close, +so nothing is lost partway through. + +That wait has no bound, and two ordinary situations never satisfy it. A process +that traps `SIGTERM` to run its own cleanup can simply decline to exit. A +descendant that inherited stdout or stderr can hold them open long after the +direct command is gone. In either case the scope stays open forever. + +`Process` implements [`@effectionx/forceable`][forceable], so wrap it in +`withForce()` to put a deadline on that wait: + +```typescript +import { main, sleep, suspend } from "effection"; +import { withForce } from "@effectionx/forceable"; +import { daemon } from "@effectionx/process"; + +await main(function* () { + let server = yield* withForce(daemon("node server.js"), function* (force) { + yield* sleep(10_000); + force("server did not exit within 10s of SIGTERM"); + }); + + yield* suspend(); +}); +``` + +The policy runs alongside the cooperative shutdown rather than replacing it. If +the process exits and its stdio closes in time, the policy is cancelled where it +stands and nothing is forced. If it does not, `force` escalates: `SIGKILL` to +the process group on POSIX, `taskkill /T /F` on Windows. Both reach descendants, +which is what makes them effective against the inherited-stdio case. + +A policy is an operation, so it can wait on application state instead of a +clock — whether a queue has drained, whether a health endpoint has gone quiet, +whether this deploy is allowed to take its time: + +```typescript +let server = yield* withForce(daemon("node server.js"), function* (force) { + let drain = yield* DrainState.expect(); + yield* drain.abandoned; + force("drain abandoned by operator"); +}); +``` + +Forcing skips whatever the process would have done on its way out — flushing +buffers, removing lock files, acknowledging in-flight work. Durable cleanup for +a process that might be forced has to be owned by something outside it. + +[forceable]: ../forceable/README.md + ## Options The `exec()` and `daemon()` functions accept an options object: @@ -224,5 +279,9 @@ interface Process { // Wait for successful completion (throws on non-zero exit) expect(): Operation; + + // Abandon cooperative shutdown and terminate the process tree. + // Called for you by withForce(); see Shutdown above. + [force](reason?: string): void; } ``` diff --git a/process/package.json b/process/package.json index 132760fa..5bbf931a 100644 --- a/process/package.json +++ b/process/package.json @@ -1,7 +1,7 @@ { "name": "@effectionx/process", "description": "Spawn and manage child processes with structured concurrency", - "version": "0.8.2", + "version": "0.9.0", "keywords": ["process"], "type": "module", "main": "./dist/mod.js", From 8c1921a4f006f19960be3293f5eecf8f8053c5df Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 19:31:32 -0400 Subject: [PATCH 11/12] =?UTF-8?q?=E2=9C=85=20Address=20review:=20determini?= =?UTF-8?q?stic=20tests=20and=20an=20honest=20force=20contract?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The withForce tests waited on the clock. Worse, the worker one allocated the shared spin flag and never read it, so on a loaded machine the halt could run against a worker that was still cooperative and the test would pass without forcing anything. Wait on the flag with when(), and assert that forcing actually happened rather than only that the halt was quiet. Give the stubborn fixture a trigger instead of a timer so every case is caused rather than timed. The suite drops from half a second to 8ms. The force contract also claimed ForcedTerminationError is reported to whoever awaits the resource. Nothing usually is: forcing happens during teardown, so on halt the awaiter is cancelled first and on normal exit the body has already returned. Say that, and point at the policy as the thing that knows. Also declare files for the new package, define the reason the README example logs, and drop the reference to a shutdown option this branch removes. --- forceable/README.md | 7 ++++ forceable/forceable.test.ts | 66 ++++++++++++++++++------------------- forceable/forceable.ts | 16 ++++++--- forceable/package.json | 1 + worker/worker-force.test.ts | 45 +++++++++++++++++++------ worker/worker.ts | 3 +- 6 files changed, 89 insertions(+), 49 deletions(-) diff --git a/forceable/README.md b/forceable/README.md index 86126c4b..94cdb124 100644 --- a/forceable/README.md +++ b/forceable/README.md @@ -87,12 +87,19 @@ force, so it is the natural place to log, count, or alert. ```ts function* (force) { + let reason = `worker did not close within ${grace}ms`; yield* sleep(grace); logger.warn({ reason }, "forced teardown"); force(reason); } ``` +The reason is recorded on a `ForcedTerminationError` that settles the +resource's outcome, but do not rely on reading it there: forcing happens +during teardown, so on halt anyone awaiting the resource is cancelled before +the rejection lands, and on a normal exit the scope body has already returned. +The policy is what knows. + Raising from inside a policy is **not** a reliable way to make forcing loud. Whether the error escapes is a race against how many turns the resource's own teardown needs after being forced. diff --git a/forceable/forceable.test.ts b/forceable/forceable.test.ts index ddd161dd..4bcfbe0d 100644 --- a/forceable/forceable.test.ts +++ b/forceable/forceable.test.ts @@ -4,30 +4,23 @@ import { ensure, resource, scoped, - sleep, spawn, suspend, withResolvers, } from "effection"; import { expect } from "expect"; -import { - ForcedTerminationError, - type Forceable, - force, - withForce, -} from "./forceable.ts"; - -interface Stubborn extends Forceable { - /** how the teardown ended, readable after the scope is gone */ - report(): string; -} +import { type Forceable, force, withForce } from "./forceable.ts"; /** - * A resource whose graceful teardown takes `closeAfter` milliseconds and which - * can be cut short. Stands in for anything holding an operating system handle. + * A resource that will not finish its graceful teardown unless something tells + * it to. `onTeardown` runs when teardown begins and receives the trigger, so a + * test decides whether the graceful path lands rather than having to time it. */ -function useStubborn(closeAfter: number, log: string[]): Operation { +function useStubborn( + log: string[], + onTeardown: (closeGracefully: () => void) => void = () => {}, +): Operation { return resource(function* (provide) { let settled = withResolvers(); let done = false; @@ -41,40 +34,39 @@ function useStubborn(closeAfter: number, log: string[]): Operation { }; yield* ensure(function* () { - yield* spawn(function* () { - yield* sleep(closeAfter); - finish("closed gracefully"); - }); + onTeardown(() => finish("closed gracefully")); yield* settled.operation; }); yield* provide({ [force]: (reason?: string) => finish(`forced: ${reason}`), - report: () => log.join(), }); }); } +/** Closes as soon as teardown begins, so the graceful path always wins. */ +const cooperative = (closeGracefully: () => void) => closeGracefully(); + describe("withForce", () => { - it("lets graceful teardown finish when it lands inside the deadline", function* () { + it("lets graceful teardown finish when the resource cooperates", function* () { let log: string[] = []; yield* scoped(function* () { - yield* withForce(useStubborn(10, log), function* (force) { - yield* sleep(200); - force("deadline expired"); + yield* withForce(useStubborn(log, cooperative), function* (force) { + yield* suspend(); + force("should never happen"); }); }); expect(log).toEqual(["closed gracefully"]); }); - it("cuts graceful teardown short when the deadline expires first", function* () { + it("cuts graceful teardown short when the policy forces", function* () { let log: string[] = []; yield* scoped(function* () { - yield* withForce(useStubborn(10_000, log), function* (force) { - yield* sleep(10); + // No onTeardown, so this resource never closes on its own. + yield* withForce(useStubborn(log), function* (force) { force("deadline expired"); }); }); @@ -84,38 +76,46 @@ describe("withForce", () => { it("cancels the policy so it cannot force after a graceful teardown", function* () { let log: string[] = []; - let policyFinished = false; + let policyResumed = false; yield* scoped(function* () { - yield* withForce(useStubborn(10, log), function* (force) { + yield* withForce(useStubborn(log, cooperative), function* (force) { yield* suspend(); - policyFinished = true; + policyResumed = true; force("should never happen"); }); }); expect(log).toEqual(["closed gracefully"]); - expect(policyFinished).toEqual(false); + expect(policyResumed).toEqual(false); }); it("stays quiet, so forcing does not disturb the halt", function* () { let log: string[] = []; + let acquired = withResolvers(); + let forced = false; let halted: Error | undefined; let task = yield* spawn(function* () { - yield* withForce(useStubborn(10_000, log), function* (force) { + yield* withForce(useStubborn(log), function* (force) { + forced = true; force("deadline expired"); }); + acquired.resolve(); yield* suspend(); }); - yield* sleep(50); + yield* acquired.operation; + try { yield* task.halt(); } catch (error) { halted = error as Error; } + // Forcing happened, and the halt still completed without an error. + expect(forced).toEqual(true); + expect(log).toEqual(["forced: deadline expired"]); expect(halted).toBeUndefined(); }); diff --git a/forceable/forceable.ts b/forceable/forceable.ts index 46c177f8..c861556a 100644 --- a/forceable/forceable.ts +++ b/forceable/forceable.ts @@ -19,9 +19,15 @@ import { export const force = Symbol.for("effection.force"); /** - * Reported to whoever is waiting on a resource that was torn down forcibly. - * Its graceful teardown never ran, so anything that teardown was responsible - * for is still outstanding. + * Settles the outcome of a resource that was torn down forcibly, so its result + * records a failure rather than a clean finish. Its graceful teardown never + * ran, so anything that teardown was responsible for is still outstanding. + * + * Note that in practice nothing is usually positioned to read it. Forcing + * happens during teardown: on halt, anyone awaiting the resource is cancelled + * first, and on normal exit the scope body has already returned. Treat the + * policy that called `force` as the place that knows, not this error. Whether + * forcing should raise instead is an open question — see the README. */ export class ForcedTerminationError extends Error { override name = "ForcedTerminationError"; @@ -37,8 +43,8 @@ export interface Forceable { * Tear down immediately. Safe to call more than once and safe to call on a * resource that already finished — later calls do nothing. * - * @param reason why the graceful teardown was abandoned, for the error the - * resource reports to whoever is waiting on it + * @param reason why the graceful teardown was abandoned, recorded on the + * {@link ForcedTerminationError} the resource settles its outcome with */ [force](reason?: string): void; } diff --git a/forceable/package.json b/forceable/package.json index cb2afc61..cabfcff7 100644 --- a/forceable/package.json +++ b/forceable/package.json @@ -27,6 +27,7 @@ "url": "https://github.com/thefrontside/effectionx/issues" }, "sideEffects": false, + "files": ["dist"], "devDependencies": { "@effectionx/vitest": "workspace:*", "effection": "^4" diff --git a/worker/worker-force.test.ts b/worker/worker-force.test.ts index 94f8ea0e..77a0369f 100644 --- a/worker/worker-force.test.ts +++ b/worker/worker-force.test.ts @@ -1,38 +1,63 @@ +import { when } from "@effectionx/converge"; +import { withForce } from "@effectionx/forceable"; import { describe, it } from "@effectionx/vitest"; -import { scoped, sleep, spawn, suspend } from "effection"; +import { type Operation, spawn, suspend } from "effection"; import { expect } from "expect"; -import { ForcedTerminationError, withForce } from "@effectionx/forceable"; import { useWorker } from "./worker.ts"; -/** A worker that spins forever and never services the close message. */ +/** + * A worker that spins forever and never services the close message, together + * with the shared flag it raises once it is genuinely non-cooperative. + */ function spinning() { let state = new Int32Array(new SharedArrayBuffer(4)); - return useWorker(import.meta.resolve("./test-assets/cpu-bound-worker.ts"), { - type: "module", - data: state.buffer, - }); + let worker = useWorker( + import.meta.resolve("./test-assets/cpu-bound-worker.ts"), + { type: "module", data: state.buffer }, + ); + return { state, worker }; +} + +/** Resolves once the worker has entered its spin loop. */ +function* spinLoopEntered(state: Int32Array): Operation { + yield* when( + function* () { + if (Atomics.load(state, 0) !== 1) { + throw new Error("worker has not started spinning"); + } + }, + { timeout: 10_000 }, + ); } describe("withForce", () => { it("stays quiet by default, so forcing does not disturb the halt", function* () { - expect.assertions(1); + expect.assertions(2); + let { state, worker } = spinning(); + let forced: string | undefined; let halted: Error | undefined; let task = yield* spawn(function* () { - yield* withForce(spinning(), function* (force) { + yield* withForce(worker, function* (force) { + forced = "cpu bound"; force("cpu bound"); }); yield* suspend(); }); - yield* sleep(200); + // Only once the worker is actually spinning is its refusal to close the + // thing this halt has to overcome. + yield* spinLoopEntered(state); + try { yield* task.halt(); } catch (error) { halted = error as Error; } + // Forcing happened, and it happened silently. + expect(forced).toEqual("cpu bound"); expect(halted).toBeUndefined(); }); }); diff --git a/worker/worker.ts b/worker/worker.ts index a7facdaf..1240ba18 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -122,7 +122,8 @@ export interface UseWorkerOptions extends WorkerOptions { * ``` * * @param url URL or string of script - * @param options Worker construction and shutdown options + * @param options Worker construction options. Shutdown is graceful; wrap + * the resource in `withForce()` from `@effectionx/forceable` to bound it. * @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 From f52d333a185a4f26633796db171c4b4d96ae3ec5 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:04:08 -0400 Subject: [PATCH 12/12] =?UTF-8?q?=F0=9F=90=9B=20Refuse=20to=20force=20a=20?= =?UTF-8?q?resource=20that=20already=20finished?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit exec()'s force implementation signalled before checking whether the process had settled, so a late call could send SIGKILL to a process group id the operating system was free to have reused. Track settlement at every resolution site and return early, which is what the Forceable contract already promised. Also tighten the tests that were passing for the wrong reasons. The cancellation test closed the resource gracefully without waiting for the policy to suspend, so it could pass without cancelling anything. The process tests slept to decide when the child was ready, which let SIGTERM arrive before the fixture installed its handlers, and then the child died of graceful shutdown rather than of forcing. Wait on the fixture's own readiness line and on the process actually being reaped. --- forceable/forceable.test.ts | 71 ++++++++++++++++------ pnpm-lock.yaml | 3 + process/package.json | 3 +- process/src/exec/posix.ts | 20 ++++++- process/src/exec/win32.ts | 21 ++++++- process/test/fixtures/cooperative.ts | 9 +++ process/test/force.test.ts | 88 ++++++++++++++++++++-------- process/tsconfig.json | 3 + 8 files changed, 168 insertions(+), 50 deletions(-) create mode 100644 process/test/fixtures/cooperative.ts diff --git a/forceable/forceable.test.ts b/forceable/forceable.test.ts index 4bcfbe0d..34c86ba3 100644 --- a/forceable/forceable.test.ts +++ b/forceable/forceable.test.ts @@ -10,7 +10,17 @@ import { } from "effection"; import { expect } from "expect"; -import { type Forceable, force, withForce } from "./forceable.ts"; +import { + ForcedTerminationError, + type Forceable, + force, + withForce, +} from "./forceable.ts"; + +interface Stubborn extends Forceable { + /** the error the resource settled with, readable after the scope is gone */ + outcome(): Error | undefined; +} /** * A resource that will not finish its graceful teardown unless something tells @@ -19,33 +29,41 @@ import { type Forceable, force, withForce } from "./forceable.ts"; */ function useStubborn( log: string[], - onTeardown: (closeGracefully: () => void) => void = () => {}, -): Operation { - return resource(function* (provide) { - let settled = withResolvers(); - let done = false; + onTeardown: ( + closeGracefully: () => void, + ) => Operation = function* () {}, +): Operation { + let settled = withResolvers(); + let done = false; + let failure: Error | undefined; - const finish = (how: string) => { + return resource(function* (provide) { + const finish = (how: string, error?: Error) => { if (!done) { done = true; + failure = error; log.push(how); settled.resolve(); } }; yield* ensure(function* () { - onTeardown(() => finish("closed gracefully")); + yield* onTeardown(() => finish("closed gracefully")); yield* settled.operation; }); yield* provide({ - [force]: (reason?: string) => finish(`forced: ${reason}`), + [force]: (reason?: string) => + finish(`forced: ${reason}`, new ForcedTerminationError(reason)), + outcome: () => failure, }); }); } /** Closes as soon as teardown begins, so the graceful path always wins. */ -const cooperative = (closeGracefully: () => void) => closeGracefully(); +function* cooperative(closeGracefully: () => void): Operation { + closeGracefully(); +} describe("withForce", () => { it("lets graceful teardown finish when the resource cooperates", function* () { @@ -63,27 +81,41 @@ describe("withForce", () => { it("cuts graceful teardown short when the policy forces", function* () { let log: string[] = []; + let stubborn: Stubborn | undefined; yield* scoped(function* () { // No onTeardown, so this resource never closes on its own. - yield* withForce(useStubborn(log), function* (force) { + stubborn = yield* withForce(useStubborn(log), function* (force) { force("deadline expired"); }); }); expect(log).toEqual(["forced: deadline expired"]); + // The resource settled as a failure rather than a clean finish. + expect(stubborn?.outcome()).toBeInstanceOf(ForcedTerminationError); + expect(stubborn?.outcome()?.message).toEqual("deadline expired"); }); it("cancels the policy so it cannot force after a graceful teardown", function* () { let log: string[] = []; let policyResumed = false; + let policyReady = withResolvers(); yield* scoped(function* () { - yield* withForce(useStubborn(log, cooperative), function* (force) { - yield* suspend(); - policyResumed = true; - force("should never happen"); - }); + yield* withForce( + // Hold the graceful close until the policy is actually suspended. + // Closing sooner would let this pass without cancelling anything. + useStubborn(log, function* (closeGracefully) { + yield* policyReady.operation; + closeGracefully(); + }), + function* (force) { + policyReady.resolve(); + yield* suspend(); + policyResumed = true; + force("should never happen"); + }, + ); }); expect(log).toEqual(["closed gracefully"]); @@ -95,9 +127,10 @@ describe("withForce", () => { let acquired = withResolvers(); let forced = false; let halted: Error | undefined; + let stubborn: Stubborn | undefined; let task = yield* spawn(function* () { - yield* withForce(useStubborn(log), function* (force) { + stubborn = yield* withForce(useStubborn(log), function* (force) { forced = true; force("deadline expired"); }); @@ -113,9 +146,9 @@ describe("withForce", () => { halted = error as Error; } - // Forcing happened, and the halt still completed without an error. + // The resource failed, and the halt that caused it still completed cleanly. expect(forced).toEqual(true); - expect(log).toEqual(["forced: deadline expired"]); + expect(stubborn?.outcome()).toBeInstanceOf(ForcedTerminationError); expect(halted).toBeUndefined(); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b0e12437..7458d780 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -233,6 +233,9 @@ importers: specifier: ^3.0.1 version: 3.0.1 devDependencies: + '@effectionx/converge': + specifier: workspace:* + version: link:../converge '@effectionx/stream-helpers': specifier: workspace:* version: link:../stream-helpers diff --git a/process/package.json b/process/package.json index 5bbf931a..66df0d34 100644 --- a/process/package.json +++ b/process/package.json @@ -38,8 +38,9 @@ "shellwords-ts": "^3.0.1" }, "devDependencies": { - "@effectionx/vitest": "workspace:*", + "@effectionx/converge": "workspace:*", "@effectionx/stream-helpers": "workspace:*", + "@effectionx/vitest": "workspace:*", "@types/cross-spawn": "^6", "effection": "^4" } diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index 1c15f672..98e68896 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -33,6 +33,14 @@ export function* createPosixProcess( options: ExecOptions, ): Operation { let processResult = withResolvers>(); + // Tracked so that [force] cannot signal a pid the OS may already have reused. + let settled = false; + const settle = (result: Result) => { + if (!settled) { + settled = true; + processResult.resolve(result); + } + }; const evalScope = yield* useEvalScope(); const result = yield* evalScope.eval(function* () { // Killing all child processes started by this command is surprisingly @@ -99,12 +107,12 @@ export function* createPosixProcess( yield* spawn(function* trapError() { let [error] = yield* once<[Error]>(childProcess, "error"); - processResult.resolve(Err(error)); + settle(Err(error)); }); yield* spawn(function* () { let value = yield* once(childProcess, "close"); - processResult.resolve(Ok(value)); + settle(Ok(value)); }); function* join() { @@ -141,6 +149,12 @@ export function* createPosixProcess( return { pid: pid as number, [force](reason?: string) { + // Forcing a resource that already finished must do nothing. Signalling + // anyway would address a process group id the OS is free to have + // reused, killing something unrelated. + if (settled) { + return; + } // SIGKILL cannot be trapped, and addressing the group reaches // descendants that are holding the inherited stdio open. try { @@ -150,7 +164,7 @@ export function* createPosixProcess( } catch (_e) { // already gone } - processResult.resolve(Err(new ForcedTerminationError(reason))); + settle(Err(new ForcedTerminationError(reason))); }, *around( ...args: Parameters diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts index bc9755e4..04303c0f 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -48,6 +48,14 @@ export function* createWin32Process( options: ExecOptions, ): Operation { let processResult = withResolvers>(); + // Tracked so that [force] cannot kill a pid the OS may already have reused. + let settled = false; + const settle = (result: Result) => { + if (!settled) { + settled = true; + processResult.resolve(result); + } + }; const evalScope = yield* useEvalScope(); const result = yield* evalScope.eval(function* () { // Windows-specific process spawning with different options than POSIX @@ -117,7 +125,7 @@ export function* createWin32Process( yield* spawn(function* trapError() { const [error] = yield* once(childProcess, "error"); - processResult.resolve(Err(error)); + settle(Err(error)); }); yield* spawn(function* () { @@ -126,7 +134,7 @@ export function* createWin32Process( // win32 is more sensitive to graceful shutdown timing that it is // worth waiting for stdout and stderr to close before resolving the process result yield* all([io.stdoutDone.operation, io.stderrDone.operation]); - processResult.resolve(Ok(value)); + settle(Ok(value)); }); function* join() { @@ -161,10 +169,13 @@ export function* createWin32Process( let forced = withResolvers(); yield* spawn(function* () { let reason = yield* forced.operation; + if (settled) { + return; + } if (pid) { yield* killTree(pid); } - processResult.resolve(Err(new ForcedTerminationError(reason))); + settle(Err(new ForcedTerminationError(reason))); }); yield* ensure(function* () { @@ -205,6 +216,10 @@ export function* createWin32Process( return { pid: pid as number, [force](reason?: string) { + // Forcing a resource that already finished must do nothing. + if (settled) { + return; + } forced.resolve(reason); }, *around( diff --git a/process/test/fixtures/cooperative.ts b/process/test/fixtures/cooperative.ts new file mode 100644 index 00000000..91c7ee53 --- /dev/null +++ b/process/test/fixtures/cooperative.ts @@ -0,0 +1,9 @@ +import process from "node:process"; + +// Exits on the first SIGTERM, so graceful shutdown always lands. +process.on("SIGTERM", () => { + process.exit(0); +}); + +console.log("ready"); +setInterval(() => {}, 1_000); diff --git a/process/test/force.test.ts b/process/test/force.test.ts index 0f3e20bc..3a542078 100644 --- a/process/test/force.test.ts +++ b/process/test/force.test.ts @@ -1,73 +1,113 @@ import process from "node:process"; +import { when } from "@effectionx/converge"; import { withForce } from "@effectionx/forceable"; +import { lines } from "@effectionx/stream-helpers"; import { describe, it } from "@effectionx/vitest"; -import { type Task, sleep, spawn, suspend } from "effection"; +import { + type Operation, + type Task, + spawn, + suspend, + withResolvers, +} from "effection"; import { expect } from "expect"; -import { exec } from "../mod.ts"; +import { type Process, exec } from "../mod.ts"; +import { expectMatch } from "./helpers.ts"; /** * Traps SIGTERM and never exits, so graceful teardown alone would wait forever. + * It prints `ready` once its handlers are installed. */ const resistant = () => `${process.execPath} ./fixtures/shutdown-resistant.ts`; +/** + * Resolves once the fixture reports that its signal handlers are installed. + * Waiting on a clock instead would let SIGTERM arrive first, in which case the + * child dies of graceful shutdown and the test proves nothing. + */ +function* readyLine(proc: Process): Operation { + yield* expectMatch(/ready/, lines()(proc.stdout)); +} + +/** Resolves once the operating system has reaped `pid`. */ +function* reaped(pid: number): Operation { + yield* when( + function* () { + try { + process.kill(pid, 0); + } catch { + return; // ESRCH: gone + } + throw new Error(`process ${pid} is still alive`); + }, + { timeout: 10_000 }, + ); +} + describe("withForce(exec())", () => { it("bounds a process that ignores graceful shutdown", function* () { - let pid: number | undefined; + let ready = withResolvers(); + let force = withResolvers(); let task: Task = yield* spawn(function* () { let proc = yield* withForce( exec(resistant(), { cwd: import.meta.dirname }), - function* (force) { - yield* sleep(100); - force("ignored SIGTERM"); + function* (force$) { + // Forcing is driven by the test rather than by a delay. + yield* force.operation; + force$("ignored SIGTERM"); }, ); - pid = proc.pid; + yield* spawn(function* () { + yield* readyLine(proc); + ready.resolve(proc.pid); + }); yield* suspend(); }); - // Give the child time to install its signal handlers. - yield* sleep(500); - expect(pid).toBeDefined(); + // Only once the handlers are installed is SIGTERM something the child can + // actually ignore. + let pid = yield* ready.operation; + force.resolve(); // Without withForce this halt never returns: the child traps SIGTERM and // teardown waits on stdio that never closes. yield* task.halt(); - // SIGKILL went to the group, so the process is actually gone rather than - // merely abandoned. kill(pid, 0) throws ESRCH once it has been reaped. - yield* sleep(100); - let alive = true; - try { - process.kill(pid as number, 0); - } catch { - alive = false; - } - expect(alive).toBe(false); + // SIGKILL went to the group, so the process is genuinely gone rather than + // merely abandoned. + yield* reaped(pid); }); it("leaves a cooperative process to shut down gracefully", function* () { + let ready = withResolvers(); let forced = false; let task: Task = yield* spawn(function* () { - yield* withForce( - exec(`${process.execPath} -e "setInterval(() => {}, 1000)"`, { + let proc = yield* withForce( + exec(`${process.execPath} ./fixtures/cooperative.ts`, { cwd: import.meta.dirname, }), function* (force) { - yield* sleep(10_000); + // Never fires on its own; only a test-owned signal would force. + yield* suspend(); forced = true; force("should not happen"); }, ); + yield* spawn(function* () { + yield* readyLine(proc); + ready.resolve(proc.pid); + }); yield* suspend(); }); - yield* sleep(300); + let pid = yield* ready.operation; yield* task.halt(); // SIGTERM was enough, so the policy was cancelled where it stood. expect(forced).toBe(false); + yield* reaped(pid); }); }); diff --git a/process/tsconfig.json b/process/tsconfig.json index 286a5b17..de11fb08 100644 --- a/process/tsconfig.json +++ b/process/tsconfig.json @@ -10,6 +10,9 @@ { "path": "../context-api" }, + { + "path": "../converge" + }, { "path": "../forceable" },