diff --git a/.changeset/quiet-startup-shutdown.md b/.changeset/quiet-startup-shutdown.md new file mode 100644 index 0000000..4331a6e --- /dev/null +++ b/.changeset/quiet-startup-shutdown.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Stop pending actor startup before closing its runtime. Wait for recovery cleanup even when a stop caller cancels its wait. Finish shutdown owner creation and cache publication before accepting caller cancellation. Preserve terminal lifecycle during initial startup and supervised activation. Keep synchronous host sends ready when startup completes. Preserve recovery cleanup errors and prevent recovery fallbacks from swallowing self-stop. diff --git a/AGENTS.md b/AGENTS.md index 6f2fc8d..4b9f495 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -279,6 +279,10 @@ const count = yield* actor.ask(Event.GetCount); // number ## Gotchas +- `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. +- 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. - ActorScope cleanup must match the registered actor identity before removing an ID. A later actor can reuse that ID. - `Machine.spawn` returns an **unstarted** actor — must call `yield* actor.start`. `system.spawn` auto-starts. diff --git a/src/actor.ts b/src/actor.ts index 2e4f598..0674561 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -136,7 +136,10 @@ export interface ActorRef | undefined >; - /** Stop the actor gracefully. */ + /** + * Stop startup and the active generation, then wait for cleanup. + * Caller interruption stops waiting without cancelling shutdown. + */ readonly stop: Effect.Effect; /** @@ -678,6 +681,16 @@ const resolveActorSystem = Effect.fn("effect-machine.resolveActorSystem")(functi return { system, implicitSystemScope: scope as Scope.Closeable | undefined }; }); +function activateGeneration( + lifecycle: SubscriptionRef.SubscriptionRef>, + generation: number, +): Effect.Effect { + return SubscriptionRef.update(lifecycle, (current): ActorLifecycle => { + if (current._tag !== "Starting" || current.generation !== generation) return current; + return { _tag: "Active", generation }; + }); +} + /** * Run the supervision loop for a supervised actor. * Observes exit deferred, applies restart policy, resets cell resources on restart. @@ -767,10 +780,7 @@ const runSupervisionLoop = < yield* newRuntime.start; const restartExit = yield* Deferred.poll(newRuntime.exitDeferred); if (Option.isNone(restartExit)) { - yield* SubscriptionRef.set(cell.lifecycleRef, { - _tag: "Active", - generation: nextGeneration, - }); + yield* activateGeneration(cell.lifecycleRef, nextGeneration); } if (options.onRestart !== undefined) { @@ -1044,12 +1054,22 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < runtimeRef.current = runtime; const supervision = options.supervision; + const startupLock = yield* Semaphore.make(1); + let stopRequested = false; + let startupFiber: Fiber.Fiber | undefined; - // Build actor stop — wraps current runtime.stop with implicit system teardown. - // For supervised actors: interrupt supervisor fiber first (cancels restart/backoff), - // then stop the current runtime, then set terminal exit. - const stopActor = Effect.fn("effect-machine.actor.stop")(function* () { - // Interrupt supervisor loop first — prevents restart during/after stop + // Shutdown owns startup cancellation, generation cleanup, and implicit system teardown. + const stopActor = Effect.fn("effect-machine.actor.stop")(function* ( + starting: Fiber.Fiber | undefined, + ) { + let cleanupCause: Cause.Cause | undefined; + if (starting !== undefined) { + yield* Fiber.interrupt(starting); + const startupExit = yield* Fiber.await(starting); + if (Exit.isFailure(startupExit) && !Cause.hasInterruptsOnly(startupExit.cause)) { + cleanupCause = startupExit.cause; + } + } if (supervisorFiberRef.current !== undefined) { yield* Fiber.interrupt(supervisorFiberRef.current); } @@ -1058,20 +1078,51 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < if (currentRuntime !== undefined) { const stopExit = yield* currentRuntime.stop.pipe(Effect.exit); const currentExit = yield* Deferred.poll(currentRuntime.exitDeferred); - if (Option.isSome(currentExit)) { - runtimeExit = yield* currentExit.value; - } else if (Exit.isFailure(stopExit)) { - runtimeExit = { _tag: "Defect", cause: stopExit.cause, phase: "cleanup" }; - } - yield* completeTerminal(runtimeExit); + if (Option.isSome(currentExit)) runtimeExit = yield* currentExit.value; if (Exit.isFailure(stopExit)) { - return yield* Effect.failCause(stopExit.cause).pipe(Effect.orDie); + if (cleanupCause === undefined) cleanupCause = stopExit.cause; + else cleanupCause = Cause.combine(cleanupCause)(stopExit.cause); + } else if (cleanupCause !== undefined && runtimeExit._tag === "Defect") { + cleanupCause = Cause.combine(cleanupCause)(runtimeExit.cause); } - } else { - yield* completeTerminal(runtimeExit); + } + if (cleanupCause !== undefined) { + runtimeExit = { _tag: "Defect", cause: cleanupCause, phase: "cleanup" }; + } + yield* completeTerminal(runtimeExit); + if (cleanupCause !== undefined) { + // @effect-diagnostics-next-line anyUnknownInErrorContext:off -- Rethrow combined cleanup causes at the actor boundary. + return yield* Effect.failCause(cleanupCause).pipe(Effect.orDie); } }); - const stop = stopActor().pipe(Effect.provide(serviceContext), Effect.asVoid); + // oxlint-disable-next-line effect/noPerCallCacheConstruction -- Actor allocation owns one shutdown fiber shared by all stop callers. + const shutdown = yield* Effect.cached( + startupLock.withPermit( + Effect.gen(function* () { + stopRequested = true; + let pendingStartup: Fiber.Fiber | undefined; + if (startupFiber?.pollUnsafe() === undefined) pendingStartup = startupFiber; + return yield* stopActor(pendingStartup).pipe( + Effect.provide(serviceContext), + Effect.forkDetach, + ); + }), + ), + ); + const stop = Effect.withFiber((caller) => + shutdown.pipe( + // Cache publication must finish before a caller can cancel its wait. + Effect.uninterruptible, + Effect.flatMap((owner) => { + if (startupFiber?.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)); + } + return Fiber.join(owner); + }), + ), + ); // Track whether hydrate was provided — skip recovery when hydrated const isHydrated = options.hydrated === true; @@ -1090,6 +1141,7 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < generation: generation.current, machineInitial: options.machineInitial, }); + if (stopRequested) return yield* Effect.interrupt; if (Option.isSome(resolved)) { // Update cell stateRef yield* SubscriptionRef.set(stateRef, resolved.value); @@ -1109,6 +1161,8 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < timestamp, })); + if (stopRequested) return yield* Effect.interrupt; + // Arm supervisor (moved from allocate → start) if (supervision !== undefined) { supervisorFiberRef.current = yield* Effect.forkDetach( @@ -1144,16 +1198,27 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < ); const currentExit = yield* Deferred.poll(currentRuntime.exitDeferred); if (Option.isNone(currentExit)) { - yield* SubscriptionRef.set(lifecycleRef, { - _tag: "Active", - generation: generation.current, - }); + yield* activateGeneration(lifecycleRef, generation.current); } } }); // oxlint-disable-next-line effect/noPerCallCacheConstruction -- Actor allocation owns one startup result for this actor. const start = yield* Effect.cached( - startActor().pipe(Effect.provide(serviceContext), Effect.asVoid), + startupLock + .withPermit( + Effect.gen(function* () { + if (stopRequested) return undefined; + startupFiber = yield* startActor().pipe(Effect.forkChild({ startImmediately: true })); + return startupFiber; + }), + ) + .pipe( + Effect.flatMap((starting) => { + if (starting === undefined) return Effect.void; + return Fiber.join(starting); + }), + Effect.provide(serviceContext), + ), ); return buildActorRefCore(cell, stop, start, serviceContext); diff --git a/src/internal/runtime.ts b/src/internal/runtime.ts index 42af478..34cfced 100644 --- a/src/internal/runtime.ts +++ b/src/internal/runtime.ts @@ -609,7 +609,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function deferredReplyRef, lifecycle, fork, - ).pipe(Effect.provide(services), Effect.forkDetach); + ).pipe(Effect.provide(services), Effect.forkDetach({ startImmediately: true })); loopFiberRef.current = loopFiber; // Background defect observer: Fiber.await each background fiber. diff --git a/test/actor.test.ts b/test/actor.test.ts index 03118b8..d912cc8 100644 --- a/test/actor.test.ts +++ b/test/actor.test.ts @@ -5,6 +5,7 @@ import { Effect, Fiber, Layer, + Queue, Ref, Schema, Context, @@ -568,33 +569,20 @@ describe("ActorRef", () => { const actor = yield* Machine.spawn(machine, { id: "test" }); yield* actor.start; - const tags: string[] = []; - - // Start collecting changes in background + const tags = yield* Queue.unbounded(); yield* Effect.forkChild( SubscriptionRef.changes(actor.state).pipe( - Stream.take(3), - Stream.tap((s) => - Effect.sync(() => { - tags.push(s._tag); - }), - ), - Stream.runDrain, + Stream.take(4), + Stream.runForEach((state) => Queue.offer(tags, state._tag)), ), ); - - // Make transitions + expect(yield* Queue.take(tags)).toBe("Idle"); yield* actor.send(TestEvent.Start({ value: 1 })); - yield* yieldFibers; + expect(yield* Queue.take(tags)).toBe("Loading"); yield* actor.send(TestEvent.Complete); - yield* yieldFibers; + expect(yield* Queue.take(tags)).toBe("Active"); yield* actor.send(TestEvent.Stop); - yield* yieldFibers; - - // Should have captured the transitions - expect(tags).toContain("Loading"); - expect(tags).toContain("Active"); - expect(tags).toContain("Done"); + expect(yield* Queue.take(tags)).toBe("Done"); }).pipe(Effect.provide(ActorSystemDefault)), ); }); diff --git a/test/stop-completion.test.ts b/test/stop-completion.test.ts index b6c926f..9034ca2 100644 --- a/test/stop-completion.test.ts +++ b/test/stop-completion.test.ts @@ -1,6 +1,6 @@ // @effect-diagnostics strictEffectProvide:off - tests are entry points // @effect-diagnostics anyUnknownInErrorContext:off -import { Cause, Deferred, Effect, Fiber, Option, Schema, SubscriptionRef } from "effect"; +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"; @@ -9,6 +9,333 @@ const LifecycleState = State({ Active: {} }); const LifecycleEvent = Event({ Ping: {} }); describe("actor stop completion", () => { + it.scopedLive("can stop after cancellation interrupts recovery self-stop", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.none()), + ), + }, + }, + }), + ); + yield* Deferred.succeed(stop, actor.stop); + const starting = yield* actor.start.pipe(Effect.forkScoped({ startImmediately: true })); + yield* Fiber.interrupt(starting); + const stopped = yield* actor.stop.pipe(Effect.exit); + expect(stopped._tag).toBe("Success"); + expect(actor.client.getLifecycle()._tag).toBe("Stopped"); + yield* actor.stop; + expect((yield* actor.awaitExit)._tag).toBe("Stopped"); + }).pipe(Effect.timeout("2 seconds")), + ); + + it.scopedLive("waits for an uninterruptible self-stop recovery to finish", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const stoppedInside = yield* Deferred.make(); + const release = yield* Deferred.make(); + let backgroundStarts = 0; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).background(() => + Effect.sync(() => { + backgroundStarts++; + }), + ); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.andThen(Deferred.succeed(stoppedInside, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.as(Option.some(LifecycleState.Active)), + Effect.uninterruptible, + ), + }, + }, + }), + ); + yield* Deferred.succeed(stop, actor.stop); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(stoppedInside); + const stopping = yield* actor.stop.pipe(Effect.forkScoped({ startImmediately: true })); + const earlyExit = stopping.pollUnsafe(); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(stopping); + expect(earlyExit).toBeUndefined(); + expect(Exit.hasInterrupts(yield* Fiber.await(starting))).toBe(true); + expect(backgroundStarts).toBe(0); + expect((yield* actor.awaitExit)._tag).toBe("Stopped"); + }), + ); + + it.scopedLive("does not resume when recovery catches a self-stop cause", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + let backgroundStarts = 0; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).background(() => + Effect.sync(() => { + backgroundStarts++; + }), + ); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.some(LifecycleState.Active)), + Effect.catchCause(() => Effect.succeedSome(LifecycleState.Active)), + ), + }, + }, + }), + ); + yield* Deferred.succeed(stop, actor.stop); + yield* actor.start.pipe(Effect.exit); + yield* actor.awaitExit; + expect(backgroundStarts).toBe(0); + }), + ); + + it.scopedLive("keeps external stop callers waiting for self-stop recovery cleanup", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const cleaning = yield* Deferred.make(); + const release = yield* Deferred.make(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.none()), + Effect.ensuring( + Deferred.succeed(cleaning, undefined).pipe( + Effect.andThen(Deferred.await(release)), + ), + ), + ), + }, + }, + }), + ); + yield* Deferred.succeed(stop, actor.stop); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(cleaning); + const stopping = yield* actor.stop.pipe(Effect.forkScoped({ startImmediately: true })); + const earlyExit = stopping.pollUnsafe(); + yield* Deferred.succeed(release, undefined); + yield* Fiber.await(starting); + yield* Fiber.join(stopping); + expect(earlyExit).toBeUndefined(); + const terminal = yield* actor.awaitExit; + expect(actor.client.getLifecycle()).toBe(terminal); + }), + ); + + it.scopedLive("reports recovery cleanup defects on every stop", () => + Effect.gen(function* () { + for (const selfStop of [false, true]) { + const entered = yield* Deferred.make(); + const stop = yield* Deferred.make>(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }); + const actor = yield* Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.succeed(entered, undefined).pipe( + Effect.andThen( + Effect.suspend(() => { + if (selfStop) + return Deferred.await(stop).pipe(Effect.flatMap((stopActor) => stopActor)); + return Effect.never; + }), + ), + Effect.andThen(Effect.never), + Effect.ensuring(Effect.die("recovery cleanup defect")), + ), + }, + }, + }); + yield* Deferred.succeed(stop, actor.stop); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const first = yield* actor.stop.pipe(Effect.exit); + const second = yield* actor.stop.pipe(Effect.exit); + yield* Fiber.await(starting); + expect(Exit.isFailure(first)).toBe(true); + expect(Exit.isFailure(second)).toBe(true); + if (Exit.isFailure(first)) + expect(Cause.pretty(first.cause)).toContain("recovery cleanup defect"); + if (Exit.isFailure(second)) + expect(Cause.pretty(second.cause)).toContain("recovery cleanup defect"); + const terminal = yield* actor.awaitExit; + expect(terminal._tag).toBe("Defect"); + if (terminal._tag === "Defect") expect(terminal.phase).toBe("cleanup"); + } + }), + ); + + it.scopedLive("reports Stopped when stop cancels an initial transition", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).immediate(LifecycleState.Active, () => + Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)), + ); + const actor = yield* Machine.scoped(Machine.spawn(machine)); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + yield* actor.stop; + expect((yield* actor.awaitExit)._tag).toBe("Stopped"); + expect(Exit.hasInterrupts(yield* Fiber.await(starting))).toBe(true); + }), + ); + + it.scopedLive("waits for recovery cleanup after the stop caller is interrupted", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const cleaning = yield* Deferred.make(); + const release = yield* Deferred.make(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Effect.never), + Effect.ensuring( + Deferred.succeed(cleaning, undefined).pipe( + Effect.andThen(Deferred.await(release)), + ), + ), + ), + }, + }, + }), + ); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const stopping = yield* actor.stop.pipe(Effect.forkScoped); + yield* Deferred.await(cleaning); + expect(stopping.pollUnsafe()).toBeUndefined(); + yield* Fiber.interrupt(stopping); + expect(starting.pollUnsafe()).toBeUndefined(); + yield* Deferred.succeed(release, undefined); + yield* actor.stop; + expect(Exit.hasInterrupts(yield* Fiber.await(starting))).toBe(true); + expect(actor.client.getLifecycle()._tag).toBe("Stopped"); + }), + ); + + it.scopedLive("can stop itself from recovery without resuming startup", () => + Effect.gen(function* () { + const stop = yield* Deferred.make>(); + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.await(stop).pipe( + Effect.flatMap((stopActor) => stopActor), + Effect.as(Option.some(LifecycleState.Active)), + ), + }, + }, + }), + ); + yield* Deferred.succeed(stop, actor.stop); + expect(Exit.hasInterrupts(yield* actor.start.pipe(Effect.exit))).toBe(true); + const terminal = yield* actor.awaitExit; + expect(actor.client.getLifecycle()).toBe(terminal); + expect(actor.client.getLifecycle()._tag).toBe("Stopped"); + }), + ); + + it.scopedLive("does not resume startup after stop during recovery", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + let started = 0; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Active, + }).background(() => + Effect.sync(() => { + started++; + }), + ); + const actor = yield* Machine.scoped( + Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as(Option.some(LifecycleState.Active)), + ), + }, + }, + }), + ); + const starting = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + yield* actor.stop; + const terminal = yield* actor.awaitExit; + yield* Deferred.succeed(release, undefined); + yield* Fiber.await(starting); + expect(actor.client.getLifecycle()).toBe(terminal); + expect(started).toBe(0); + }), + ); + it.scopedLive("stop returns with a terminal public lifecycle", () => Effect.gen(function* () { const machine = Machine.make({