Stop pending actor startup before shutdown

.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.

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.

src/actor.ts

@@ -136,7 +136,10 @@ export interface ActorRef<State extends { readonly _tag: string }, Event, Output
     TransitionInfo<State, Event> | 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<void>;
 
   /**
@@ -678,6 +681,16 @@ const resolveActorSystem = Effect.fn("effect-machine.resolveActorSystem")(functi
   return { system, implicitSystemScope: scope as Scope.Closeable | undefined };
 });
 
+function activateGeneration<S extends AnyState, O>(
+  lifecycle: SubscriptionRef.SubscriptionRef<ActorLifecycle<S, O>>,
+  generation: number,
+): Effect.Effect<void> {
+  return SubscriptionRef.update(lifecycle, (current): ActorLifecycle<S, O> => {
+    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<void> | 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<void> | undefined,
+  ) {
+    let cleanupCause: Cause.Cause<unknown> | 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<void> | 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);

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.

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<string>();
         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)),
     );
   });