From c77e4340dd5be4ca8f120f72999690aecf86503c Mon Sep 17 00:00:00 2001 From: Cristian Date: Sun, 6 Sep 2026 23:05:58 +0000 Subject: [PATCH 1/2] fix: finish supervised recovery self-stop cleanup --- .changeset/plain-supervisor-stop.md | 5 ++ AGENTS.md | 1 + src/actor.ts | 7 +- test/stop-completion.test.ts | 103 +++++++++++++++++++++++++++- 4 files changed, 114 insertions(+), 2 deletions(-) create mode 100644 .changeset/plain-supervisor-stop.md diff --git a/.changeset/plain-supervisor-stop.md b/.changeset/plain-supervisor-stop.md new file mode 100644 index 0000000..c7e46d3 --- /dev/null +++ b/.changeset/plain-supervisor-stop.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Allow an actor to stop itself from protected supervised recovery. Wait for the supervisor to finish cleanup and report its cleanup defects to every stop caller. diff --git a/AGENTS.md b/AGENTS.md index 4b9f495..d1266eb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -281,6 +281,7 @@ const count = yield* actor.ask(Event.GetCount); // number - `actor.stop` cancels pending startup and waits for recovery cleanup. Stop callers can cancel their own wait without cancelling shutdown. A stop from recovery marks that startup interrupted. Shutdown waits for its protected regions and finalizers to finish. Recovery cleanup defects reach every stop caller. - Protect both shutdown owner creation and its cache publication from interruption. Protecting only the cached body can cache cancellation when its protected region ends. +- Recovery self-stop must also identify the supervisor fiber. A protected supervisor must not join an owner that waits for that same supervisor. Preserve supervised recovery cleanup defects on stop. - The runtime event loop must be ready before startup completes. Host client sends must still commit synchronous transitions before returning. - Publish Active only while the same generation is still Starting. A terminal lifecycle must never return to Active. - `actor.start` shares one startup result across concurrent and repeated calls, including failure or interruption. Stop the actor and spawn a new actor to retry initialization. It cannot restart a terminal actor. diff --git a/src/actor.ts b/src/actor.ts index 0674561..6cd2af2 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -1072,6 +1072,11 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < } if (supervisorFiberRef.current !== undefined) { yield* Fiber.interrupt(supervisorFiberRef.current); + const supervisorExit = yield* Fiber.await(supervisorFiberRef.current); + if (Exit.isFailure(supervisorExit) && !Cause.hasInterruptsOnly(supervisorExit.cause)) { + if (cleanupCause === undefined) cleanupCause = supervisorExit.cause; + else cleanupCause = Cause.combine(cleanupCause)(supervisorExit.cause); + } } const currentRuntime = runtimeRef.current; let runtimeExit: RuntimeExit = { _tag: "Stopped" }; @@ -1114,7 +1119,7 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < // Cache publication must finish before a caller can cancel its wait. Effect.uninterruptible, Effect.flatMap((owner) => { - if (startupFiber?.id === caller.id) { + if (startupFiber?.id === caller.id || supervisorFiberRef.current?.id === caller.id) { // Mark the fiber itself. A cause-level fallback must not swallow self-stop. // Do not join self: recovery may be inside an uninterruptible region. return Effect.sync(() => caller.interruptUnsafe(caller.id)); diff --git a/test/stop-completion.test.ts b/test/stop-completion.test.ts index 9034ca2..3bb8936 100644 --- a/test/stop-completion.test.ts +++ b/test/stop-completion.test.ts @@ -3,12 +3,113 @@ import { Cause, Deferred, Effect, Exit, Fiber, Option, Schema, SubscriptionRef } from "effect"; import { describe, expect, it, yieldFibers } from "effect-bun-test"; -import { Event, Machine, State } from "../src/index.js"; +import { Event, Machine, State, Supervision } from "../src/index.js"; const LifecycleState = State({ Active: {} }); const LifecycleEvent = Event({ Ping: {} }); describe("actor stop completion", () => { + it.scopedLive("waits for supervised self-stop cleanup and preserves its defects", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const cleaning = yield* Deferred.make(); + const release = yield* Deferred.make(); + let starts = 0; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).spawn(LifecycleState.Active, () => + Effect.suspend(() => { + starts++; + return Effect.die("restart for cleanup"); + }), + ); + const actor = yield* Machine.spawn(machine, { + supervision: Supervision.restart({ maxRestarts: 1 }), + lifecycle: { + recovery: { + resolve: ({ generation }) => { + if (generation === 0) return Effect.succeedNone; + return Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.none()), + Effect.catchCause(() => Effect.succeedNone), + Effect.ensuring( + Deferred.succeed(cleaning, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen(Effect.die("supervised recovery cleanup")), + ), + ), + ); + }, + }, + }, + }); + yield* Deferred.succeed(stop, actor.stop); + yield* actor.start; + yield* Deferred.await(cleaning); + const stopping = yield* actor.stop.pipe(Effect.forkScoped({ startImmediately: true })); + yield* yieldFibers; + const earlyExit = stopping.pollUnsafe(); + yield* Deferred.succeed(release, undefined); + const stopped = yield* Fiber.await(stopping); + const repeated = yield* actor.stop.pipe(Effect.exit); + expect(earlyExit).toBeUndefined(); + for (const exit of [stopped, repeated]) { + expect(exit._tag).toBe("Failure"); + if (Exit.isFailure(exit)) { + expect(Cause.pretty(exit.cause)).toContain("supervised recovery cleanup"); + } + } + expect(starts).toBe(1); + const terminal = yield* actor.awaitExit; + expect(terminal._tag).toBe("Defect"); + if (terminal._tag === "Defect") expect(terminal.phase).toBe("cleanup"); + }).pipe(Effect.timeout("2 seconds")), + ); + + it.scopedLive("can stop itself during protected supervised recovery", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const recovering = yield* Deferred.make(); + let starts = 0; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).spawn(LifecycleState.Active, () => + Effect.suspend(() => { + starts++; + return Effect.die("restart for recovery"); + }), + ); + const actor = yield* Machine.spawn(machine, { + supervision: Supervision.restart({ maxRestarts: 1 }), + lifecycle: { + recovery: { + resolve: ({ generation }) => { + if (generation === 0) return Effect.succeedNone; + return Deferred.succeed(recovering, undefined).pipe( + Effect.andThen(Deferred.await(stop)), + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.none()), + Effect.uninterruptible, + ); + }, + }, + }, + }); + yield* Deferred.succeed(stop, actor.stop); + yield* actor.start; + yield* Deferred.await(recovering); + const stopped = yield* actor.stop.pipe(Effect.timeout("1 second"), Effect.exit); + expect(stopped._tag).toBe("Success"); + expect(starts).toBe(1); + expect((yield* actor.awaitExit)._tag).toBe("Defect"); + }).pipe(Effect.timeout("2 seconds")), + ); + it.scopedLive("can stop after cancellation interrupts recovery self-stop", () => Effect.gen(function* () { const stop = yield* Deferred.make>(); From a023ae1385865e938b61bec437146f96aad9c7c3 Mon Sep 17 00:00:00 2001 From: Cristian Date: Sun, 6 Sep 2026 23:08:16 +0000 Subject: [PATCH 2/2] fix: apply supervision budget to restarted startup failures --- .changeset/plain-supervisor-stop.md | 2 + AGENTS.md | 1 + src/actor.ts | 3 +- test/supervision.test.ts | 58 ++++++++++++++++++++++++++++- 4 files changed, 62 insertions(+), 2 deletions(-) diff --git a/.changeset/plain-supervisor-stop.md b/.changeset/plain-supervisor-stop.md index c7e46d3..0f8346e 100644 --- a/.changeset/plain-supervisor-stop.md +++ b/.changeset/plain-supervisor-stop.md @@ -3,3 +3,5 @@ --- Allow an actor to stop itself from protected supervised recovery. Wait for the supervisor to finish cleanup and report its cleanup defects to every stop caller. + +Apply the restart policy when a restarted generation fails during startup. Continue within the retry budget and complete the actor exit when that budget ends. diff --git a/AGENTS.md b/AGENTS.md index d1266eb..6947806 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -282,6 +282,7 @@ const count = yield* actor.ask(Event.GetCount); // number - `actor.stop` cancels pending startup and waits for recovery cleanup. Stop callers can cancel their own wait without cancelling shutdown. A stop from recovery marks that startup interrupted. Shutdown waits for its protected regions and finalizers to finish. Recovery cleanup defects reach every stop caller. - Protect both shutdown owner creation and its cache publication from interruption. Protecting only the cached body can cache cancellation when its protected region ends. - Recovery self-stop must also identify the supervisor fiber. A protected supervisor must not join an owner that waits for that same supervisor. Preserve supervised recovery cleanup defects on stop. +- A supervised generation can fail during `runtime.start`. Let the loop read its recorded exit and apply the restart policy. Do not let that start failure end the supervisor before it completes the actor exit. - The runtime event loop must be ready before startup completes. Host client sends must still commit synchronous transitions before returning. - Publish Active only while the same generation is still Starting. A terminal lifecycle must never return to Active. - `actor.start` shares one startup result across concurrent and repeated calls, including failure or interruption. Stop the actor and spawn a new actor to retry initialization. It cannot restart a terminal actor. diff --git a/src/actor.ts b/src/actor.ts index 6cd2af2..20f0a79 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -777,7 +777,8 @@ const runSupervisionLoop = < const newRuntime = yield* options.spawnGeneration(cell.machine); cell.runtimeRef.current = newRuntime; - yield* newRuntime.start; + // The runtime records startup failure. The next iteration applies the restart policy. + yield* newRuntime.start.pipe(Effect.ignoreCause); const restartExit = yield* Deferred.poll(newRuntime.exitDeferred); if (Option.isNone(restartExit)) { yield* activateGeneration(cell.lifecycleRef, nextGeneration); diff --git a/test/supervision.test.ts b/test/supervision.test.ts index cd581cf..bc76387 100644 --- a/test/supervision.test.ts +++ b/test/supervision.test.ts @@ -1,5 +1,5 @@ // @effect-diagnostics strictEffectProvide:off - tests are entry points -import { Deferred, Duration, Effect, Schema } from "effect"; +import { Cause, Deferred, Duration, Effect, Schema } from "effect"; import { ActorSystemDefault, @@ -34,6 +34,62 @@ const machine = Machine.make({ state: S, event: E, initial: S.Idle }) // ============================================================================ describe("supervision: restart on defect", () => { + it.scopedLive("exhausts the restart budget after repeated initial spawn defects", () => + Effect.gen(function* () { + let attempts = 0; + const failingMachine = Machine.make({ state: S, event: E, initial: S.Idle }).spawn( + S.Idle, + () => + Effect.suspend(() => { + attempts++; + return Effect.die("every initial spawn fails"); + }), + ); + const actor = yield* Machine.spawn(failingMachine, { + supervision: Supervision.restart({ maxRestarts: 2 }), + }); + yield* actor.start; + const exit = yield* actor.awaitExit.pipe(Effect.timeout("1 second")); + expect(attempts).toBe(3); + expect(exit._tag).toBe("Defect"); + if (exit._tag === "Defect") { + expect(exit.phase).toBe("initial-spawn"); + expect(Cause.pretty(exit.cause)).toContain("every initial spawn fails"); + } + expect(actor.client.getLifecycle()).toBe(exit); + yield* actor.stop; + yield* actor.stop; + }), + ); + + it.scopedLive("can recover after an initial spawn fails in a restarted generation", () => + Effect.gen(function* () { + let attempts = 0; + const ready = yield* Deferred.make(); + const recoveringMachine = Machine.make({ state: S, event: E, initial: S.Idle }) + .spawn(S.Idle, () => + Effect.suspend(() => { + attempts++; + if (attempts < 3) return Effect.die("retry initial spawn"); + return Deferred.succeed(ready, undefined); + }), + ) + .on(S.Idle, E.Finish, () => S.Done) + .final(S.Done); + const actor = yield* Machine.spawn(recoveringMachine, { + supervision: Supervision.restart({ maxRestarts: 2 }), + }); + yield* actor.start; + yield* Deferred.await(ready).pipe(Effect.timeout("1 second")); + yield* actor.call(E.Finish); + const exit = yield* actor.awaitExit; + expect(attempts).toBe(3); + expect(exit._tag).toBe("Final"); + expect(actor.client.getLifecycle()).toBe(exit); + yield* actor.stop; + }), + ); + it.scopedLive("restarts after synchronous initial state Effect defect", () => Effect.gen(function* () { const restarted = yield* Deferred.make();