From 6c19e688d0766e91c7871f0ae03c67d1c505f024 Mon Sep 17 00:00:00 2001 From: Cristian Date: Wed, 2 Sep 2026 10:07:42 +0000 Subject: [PATCH] fix(runtime): commit state before entry effects --- .changeset/commit-state-before-effects.md | 5 + src/actor.ts | 10 -- src/internal/runtime.ts | 73 +++++++-------- src/internal/transition.ts | 108 ++++++++++------------ test/eager-effect-start.test.ts | 81 +++++++++++++++- 5 files changed, 169 insertions(+), 108 deletions(-) create mode 100644 .changeset/commit-state-before-effects.md diff --git a/.changeset/commit-state-before-effects.md b/.changeset/commit-state-before-effects.md new file mode 100644 index 0000000..c1c4d2b --- /dev/null +++ b/.changeset/commit-state-before-effects.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Commit an entered state before its state Effect starts. diff --git a/src/actor.ts b/src/actor.ts index 58d0097..2d03a53 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -909,15 +909,6 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < timestamp, })); }); - const onInitialSpawnEffects: RuntimeLifecycleHooks["onInitialSpawnEffects"] = (state) => - emitWithTimestamp(inspectorValue, (timestamp) => ({ - type: "@machine.effect", - actorId: id, - generation: runtimeGeneration, - effectType: "spawn", - state, - timestamp, - })); return { onEvent, onStateChange: (result, event) => @@ -964,7 +955,6 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < })); } }), - onInitialSpawnEffects, }; }; diff --git a/src/internal/runtime.ts b/src/internal/runtime.ts index 6ab3843..529b1bb 100644 --- a/src/internal/runtime.ts +++ b/src/internal/runtime.ts @@ -34,7 +34,7 @@ import type { Machine, MachineRef } from "../machine.js"; import type { ActorRef, ActorSystemService, TransitionInfo } from "../actor.js"; import { ActorSystem as ActorSystemTag } from "../actor.js"; import type { ProcessEventHooks, ProcessEventResult } from "./transition.js"; -import { processEventCoreImmediate, runSpawnEffects, shouldPostpone } from "./transition.js"; +import { enterState, processEventCoreImmediate, shouldPostpone } from "./transition.js"; import { makeEventAdvancement } from "./event-advancement.js"; import { ActorStoppedError, NoReplyError } from "../errors.js"; import { INTERNAL_INIT_EVENT, isEffect } from "./utils.js"; @@ -159,8 +159,6 @@ export interface RuntimeLifecycleHooks { readonly onFinal?: (state: S) => Effect.Effect; /** Before stop resource cleanup — actor emits @machine.stop, settles pending replies */ readonly onShutdown?: () => Effect.Effect; - /** Before initial spawn effects — actor emits @machine.effect inspection */ - readonly onInitialSpawnEffects?: (state: S) => Effect.Effect; } // ============================================================================ @@ -324,12 +322,8 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function machine, initialState, initEvent, - self, stateScopeRef, - system, - actorId, { ...hooks, onSpawnDefect: initialSpawnDefectSignal }, - generation, ); let initialResult: ProcessEventResult; if (isEffect(initialProcessing)) { @@ -388,36 +382,29 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function // Run initial spawn effects — catch defects, tag as initial-spawn, and propagate. // For unsupervised actors this fails createActor (correct: don't register dead actors). // For supervised actors (Step 3), the supervision loop will catch and restart. - if (!initialResult.lifecycleRan && lifecycle?.onInitialSpawnEffects !== undefined) { - yield* lifecycle.onInitialSpawnEffects(stableInitialState); - } // Note: onSpawnDefect for initial spawn fibers that defect asynchronously (after forking). // If they defect later, this signals through exitDeferred and interrupts the loop. - if (!initialResult.lifecycleRan) { - yield* runSpawnEffects( - machine, - stableInitialState, - initEvent, - self, - stateScopeRef.current, - system, - actorId, - hooks?.onError, - initialSpawnDefectSignal, - generation, - ).pipe( - Effect.catchCause((cause) => - // Tag as initial-spawn defect, set exit, clean up, then propagate - Effect.gen(function* () { - yield* Ref.set(stoppedRef, true); - yield* Scope.close(stateScopeRef.current, Exit.void); - yield* Scope.close(actorScope, Exit.void); - yield* Deferred.succeed(exitDeferred, RuntimeExit.Defect(cause, "initial-spawn")); - return yield* Effect.failCause(cause); - }), - ), - ); - } + yield* enterState( + machine, + stableInitialState, + self, + stateScopeRef.current, + system, + actorId, + { ...hooks, onSpawnDefect: initialSpawnDefectSignal }, + generation, + ).pipe( + Effect.catchCause((cause) => + // Tag as initial-spawn defect, set exit, clean up, then propagate + Effect.gen(function* () { + yield* Ref.set(stoppedRef, true); + yield* Scope.close(stateScopeRef.current, Exit.void); + yield* Scope.close(actorScope, Exit.void); + yield* Deferred.succeed(exitDeferred, RuntimeExit.Defect(cause, "initial-spawn")); + return yield* Effect.failCause(cause); + }), + ), + ); // Check if initial state is final — if so, clean up and signal done if (machine._isFinal(stableInitialState._tag)) { @@ -741,12 +728,8 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function* machine, currentState, event, - self, stateScopeRef, - system, - actorId, hooks, - generation, ); let result: ProcessEventResult; if (isEffect(processing)) { @@ -766,6 +749,18 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function* event: latest.event, }); } + if (result.lifecycleRan) { + yield* enterState( + machine, + result.newState, + self, + stateScopeRef.current, + system, + actorId, + hooks, + generation, + ); + } } // Lifecycle: onStateChange (actor notifies listeners and saves durability) diff --git a/src/internal/transition.ts b/src/internal/transition.ts index 34075be..53d1e38 100644 --- a/src/internal/transition.ts +++ b/src/internal/transition.ts @@ -408,19 +408,24 @@ export const processEventCore = < generation = 0, ) => Effect.suspend(() => { - const processed = processEventCoreImmediate( - machine, - currentState, - event, - self, - stateScopeRef, - system, - actorId, - hooks, - generation, - ); - if (isEffect(processed)) return processed; - return Effect.succeed(processed); + const processed = processEventCoreImmediate(machine, currentState, event, stateScopeRef, hooks); + return Effect.gen(function* () { + let result: ProcessEventResult; + if (isEffect(processed)) result = yield* processed; + else result = processed; + if (!result.lifecycleRan) return result; + yield* enterState( + machine, + result.newState, + self, + stateScopeRef.current, + system, + actorId, + hooks, + generation, + ); + return result; + }); }); const completeProcessedEvent = < @@ -431,14 +436,9 @@ const completeProcessedEvent = < // eslint-disable-next-line @typescript-eslint/no-explicit-any machine: Machine, currentState: S, - event: E, result: ExecutedTransition, - self: MachineRef, stateScopeRef: { current: Scope.Closeable }, - system: ActorSystemService, - actorId: string, hooks?: ProcessEventHooks, - generation = 0, ): | ProcessEventResult | Effect.Effect, never, Exclude> => { @@ -497,25 +497,6 @@ const completeProcessedEvent = < ); } - // Hook: about to run spawn effects - if (hooks?.onSpawnEffect !== undefined) { - yield* hooks.onSpawnEffect(newState); - } - - // Run spawn effects for new state - const enterEvent = { _tag: INTERNAL_ENTER_EVENT } as E; - yield* runSpawnEffects( - machine, - newState, - enterEvent, - self, - stateScopeRef.current, - system, - actorId, - hooks?.onError, - hooks?.onSpawnDefect, - generation, - ); return processed; }); }; @@ -530,29 +511,14 @@ export const processEventCoreImmediate = < machine: Machine, currentState: S, event: E, - self: MachineRef, stateScopeRef: { current: Scope.Closeable }, - system: ActorSystemService, - actorId: string, hooks?: ProcessEventHooks, - generation = 0, ) => { const execution = executeTransitionImmediate(machine, currentState, event, hooks); const complete = (result: ExecutedTransition) => { const immediateCandidates = machine._findImmediateTransitions(result.newState._tag); if (immediateCandidates.length === 0) { - return completeProcessedEvent( - machine, - currentState, - event, - result, - self, - stateScopeRef, - system, - actorId, - hooks, - generation, - ); + return completeProcessedEvent(machine, currentState, result, stateScopeRef, hooks); } return Effect.gen(function* () { @@ -591,7 +557,6 @@ export const processEventCoreImmediate = < const processed = completeProcessedEvent( machine, currentState, - event, { ...result, newState: stableState, @@ -600,12 +565,8 @@ export const processEventCoreImmediate = < transition: steps.at(-1)?.transition, steps, }, - self, stateScopeRef, - system, - actorId, hooks, - generation, ); if (isEffect(processed)) return yield* processed; return processed; @@ -698,6 +659,37 @@ export const runSpawnEffects = Effect.fn("effect-machine.runSpawnEffects")(funct } }); +/** @internal */ +export const enterState = Effect.fn("effect-machine.enterState")(function* < + S extends { readonly _tag: string }, + E extends { readonly _tag: string }, + R, +>( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + machine: Machine, + state: S, + self: MachineRef, + stateScope: Scope.Closeable, + system: ActorSystemService, + actorId: string, + hooks?: ProcessEventHooks, + generation = 0, +) { + if (hooks?.onSpawnEffect !== undefined) yield* hooks.onSpawnEffect(state); + yield* runSpawnEffects( + machine, + state, + { _tag: INTERNAL_ENTER_EVENT } as E, + self, + stateScope, + system, + actorId, + hooks?.onError, + hooks?.onSpawnDefect, + generation, + ); +}); + /** * Resolve which transition should fire for a given state and event. * Uses indexed O(1) lookup. First matching transition wins. diff --git a/test/eager-effect-start.test.ts b/test/eager-effect-start.test.ts index db0d5cd..7ea04d2 100644 --- a/test/eager-effect-start.test.ts +++ b/test/eager-effect-start.test.ts @@ -1,5 +1,5 @@ // @effect-diagnostics strictEffectProvide:off - tests are entry points -import { Cause, Effect, Schema } from "effect"; +import { Cause, Effect, Schema, SubscriptionRef } from "effect"; import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry"; import { describe, expect, it } from "effect-bun-test"; @@ -10,6 +10,84 @@ const LifecycleState = State({ Idle: {}, Listening: {}, Done: {} }); const LifecycleEvent = Event({ Start: {}, Stop: {} }); describe("actor Effect eager start", () => { + it.scopedLive("commits the entered state before its Effect starts", () => + Effect.gen(function* () { + let observedState: typeof LifecycleState.Type | undefined; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, ({ self }) => + SubscriptionRef.get(self.state).pipe( + Effect.tap((state) => Effect.sync(() => (observedState = state))), + Effect.andThen(Effect.never), + ), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + yield* actor.call(LifecycleEvent.Start); + + expect(observedState).toEqual(LifecycleState.Listening); + yield* actor.stop; + }), + ); + + it.scopedLive("commits an initial immediate state before its Effect starts", () => + Effect.gen(function* () { + let observedState: typeof LifecycleState.Type | undefined; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .immediate(LifecycleState.Idle, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, ({ self }) => + SubscriptionRef.get(self.state).pipe( + Effect.tap((state) => Effect.sync(() => (observedState = state))), + Effect.andThen(Effect.never), + ), + ); + const actor = yield* Machine.spawn(machine); + + yield* actor.start; + + expect(observedState).toEqual(LifecycleState.Listening); + yield* actor.stop; + }), + ); + + it.scopedLive("commits reentry data before its Effect starts", () => + Effect.gen(function* () { + const ReentryState = State({ Active: { revision: Schema.Finite } }); + const ReentryEvent = Event({ Bump: {} }); + const observedRevisions: Array = []; + const machine = Machine.make({ + state: ReentryState, + event: ReentryEvent, + initial: ReentryState.Active({ revision: 0 }), + }) + .reenter(ReentryState.Active, ReentryEvent.Bump, ({ state }) => + ReentryState.Active.with(state, { revision: state.revision + 1 }), + ) + .spawn(ReentryState.Active, ({ self }) => + SubscriptionRef.get(self.state).pipe( + Effect.tap((state) => Effect.sync(() => observedRevisions.push(state.revision))), + Effect.andThen(Effect.never), + ), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + yield* actor.call(ReentryEvent.Bump); + + expect(observedRevisions).toEqual([0, 1]); + yield* actor.stop; + }), + ); + it.scopedLive("starts state Effect setup before state subscribers run", () => Effect.gen(function* () { const records: Array = []; @@ -161,6 +239,7 @@ describe("actor Effect eager start", () => { expect(Cause.hasInterruptsOnly(exit.cause)).toBe(false); expect(Cause.pretty(exit.cause)).toContain("spawn boom"); } + expect(yield* actor.snapshot).toEqual(LifecycleState.Listening); }), );