✨ Add @effectionx/forceable and implement it on Worker and Process

forceable/README.md

@@ -0,0 +1,105 @@
+# 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) {
+  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.

forceable/forceable.test.ts

@@ -0,0 +1,159 @@
+import { describe, it } from "@effectionx/vitest";
+import {
+  type Operation,
+  ensure,
+  resource,
+  scoped,
+  spawn,
+  suspend,
+  withResolvers,
+} from "effection";
+import { expect } from "expect";
+
+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
+ * 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(
+  log: string[],
+  onTeardown: (
+    closeGracefully: () => void,
+  ) => Operation<void> = function* () {},
+): Operation<Stubborn> {
+  let settled = withResolvers<void>();
+  let done = false;
+  let failure: Error | undefined;
+
+  return resource(function* (provide) {
+    const finish = (how: string, error?: Error) => {
+      if (!done) {
+        done = true;
+        failure = error;
+        log.push(how);
+        settled.resolve();
+      }
+    };
+
+    yield* ensure(function* () {
+      yield* onTeardown(() => finish("closed gracefully"));
+      yield* settled.operation;
+    });
+
+    yield* provide({
+      [force]: (reason?: string) =>
+        finish(`forced: ${reason}`, new ForcedTerminationError(reason)),
+      outcome: () => failure,
+    });
+  });
+}
+
+/** Closes as soon as teardown begins, so the graceful path always wins. */
+function* cooperative(closeGracefully: () => void): Operation<void> {
+  closeGracefully();
+}
+
+describe("withForce", () => {
+  it("lets graceful teardown finish when the resource cooperates", function* () {
+    let log: string[] = [];
+
+    yield* scoped(function* () {
+      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 policy forces", function* () {
+    let log: string[] = [];
+    let stubborn: Stubborn | undefined;
+
+    yield* scoped(function* () {
+      // No onTeardown, so this resource never closes on its own.
+      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<void>();
+
+    yield* scoped(function* () {
+      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"]);
+    expect(policyResumed).toEqual(false);
+  });
+
+  it("stays quiet, so forcing does not disturb the halt", function* () {
+    let log: string[] = [];
+    let acquired = withResolvers<void>();
+    let forced = false;
+    let halted: Error | undefined;
+    let stubborn: Stubborn | undefined;
+
+    let task = yield* spawn(function* () {
+      stubborn = yield* withForce(useStubborn(log), function* (force) {
+        forced = true;
+        force("deadline expired");
+      });
+      acquired.resolve();
+      yield* suspend();
+    });
+
+    yield* acquired.operation;
+
+    try {
+      yield* task.halt();
+    } catch (error) {
+      halted = error as Error;
+    }
+
+    // The resource failed, and the halt that caused it still completed cleanly.
+    expect(forced).toEqual(true);
+    expect(stubborn?.outcome()).toBeInstanceOf(ForcedTerminationError);
+    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.
+});

forceable/forceable.ts

@@ -0,0 +1,119 @@
+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");
+
+/**
+ * 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";
+
+  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, recorded on the
+   * {@link ForcedTerminationError} the resource settles its outcome with
+   */
+  [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<void>;
+
+/**
+ * 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<T extends Forceable>(
+  op: Operation<T>,
+  policy: ForcePolicy,
+): Operation<T> {
+  return resource(function* (provide) {
+    let acquired = withResolvers<T>();
+
+    // 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);
+  });
+}

forceable/mod.ts

@@ -0,0 +1 @@
+export * from "./forceable.ts";