From 69d423ccddce34e2ca7ca7264f32face253b9b41 Mon Sep 17 00:00:00 2001 From: Cristian Date: Tue, 1 Sep 2026 12:45:26 +0000 Subject: [PATCH] fix(runtime): start actor effects eagerly --- .changeset/eager-actor-effects.md | 5 + src/actor.ts | 7 +- src/internal/runtime.ts | 22 +-- src/internal/transition.ts | 8 +- test/actor.test.ts | 10 +- test/eager-effect-start.test.ts | 232 ++++++++++++++++++++++++++++++ test/inspection.test.ts | 4 +- test/supervision.test.ts | 25 ++++ 8 files changed, 296 insertions(+), 17 deletions(-) create mode 100644 .changeset/eager-actor-effects.md create mode 100644 test/eager-effect-start.test.ts diff --git a/.changeset/eager-actor-effects.md b/.changeset/eager-actor-effects.md new file mode 100644 index 0000000..dae5788 --- /dev/null +++ b/.changeset/eager-actor-effects.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Start state and background Effects eagerly so their synchronous setup completes before actor state becomes visible. A task that completes in its first Effect slice can now enqueue its result before a caller sends its next event. A suspending Inspector can delay this setup because inspection remains ordered and awaited. diff --git a/src/actor.ts b/src/actor.ts index 6073511..40b9bb2 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -1094,7 +1094,12 @@ export const createActor = Effect.fn("effect-machine.actor.spawn")(function* < // Delegate to runtime.start (forks event loop, background, spawn effects) const currentRuntime = runtimeRef.current; if (currentRuntime !== undefined) { - yield* currentRuntime.start; + yield* currentRuntime.start.pipe( + Effect.catchCause((cause) => { + if (supervision === undefined) return Effect.failCause(cause); + return Effect.void; + }), + ); const currentExit = yield* Deferred.poll(currentRuntime.exitDeferred); if (Option.isNone(currentExit)) { yield* SubscriptionRef.set(lifecycleRef, { diff --git a/src/internal/runtime.ts b/src/internal/runtime.ts index 4a428e9..cfa08f3 100644 --- a/src/internal/runtime.ts +++ b/src/internal/runtime.ts @@ -381,7 +381,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function self, system, }) - .pipe(Effect.forkIn(actorScope)); + .pipe(Effect.forkIn(actorScope, { startImmediately: true })); backgroundFibers.push(fiber); } @@ -438,10 +438,9 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function Deferred.succeed(exitDeferred, RuntimeExit.Defect(cause, "spawn")).pipe( Effect.andThen(Ref.set(stoppedRef, true)), Effect.andThen( - Effect.suspend(() => { + Effect.sync(() => { const loopFiber = loopFiberRef.current; - if (loopFiber !== undefined) return Fiber.interrupt(loopFiber); - return Effect.void; + if (loopFiber !== undefined) fork(Fiber.interrupt(loopFiber)); }), ), Effect.asVoid, @@ -725,6 +724,10 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function* Effect.gen(function* () { const event = queued.event; + if (queued._tag === "ask" && deferredReplyRef !== undefined) { + deferredReplyRef.current = queued.reply; + } + // Lifecycle: onEvent (actor emits @machine.event) if (lifecycle?.onEvent !== undefined) yield* lifecycle.onEvent(currentState, event); @@ -776,6 +779,9 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function* break; case "ask": if (result.hasReply) { + if (deferredReplyRef?.current === queued.reply) { + deferredReplyRef.current = undefined; + } const replySchema = machine._replySchema(event._tag); if (replySchema !== undefined) { const decoded = yield* Schema.decodeUnknownEffect(replySchema)(result.reply).pipe( @@ -790,10 +796,10 @@ const runtimeEventLoop = Effect.fn("effect-machine.runtime.eventLoop")(function* } else { yield* Deferred.succeed(queued.reply, result.reply); } - } else if (result.deferReply && deferredReplyRef !== undefined) { - // Handler returned Machine.deferReply() — spawn handler will call self.reply() - deferredReplyRef.current = queued.reply; - } else { + } else if (!result.deferReply) { + if (deferredReplyRef?.current === queued.reply) { + deferredReplyRef.current = undefined; + } yield* Deferred.fail( queued.reply, NoReplyError.make({ actorId, eventTag: event._tag }), diff --git a/src/internal/transition.ts b/src/internal/transition.ts index b849c68..34075be 100644 --- a/src/internal/transition.ts +++ b/src/internal/transition.ts @@ -688,7 +688,13 @@ export const runSpawnEffects = Effect.fn("effect-machine.runSpawnEffects")(funct }), ); - yield* Effect.forkScoped(effect).pipe(Effect.provideService(Scope.Scope, stateScope)); + const fiber = yield* Effect.forkScoped(effect, { startImmediately: true }).pipe( + Effect.provideService(Scope.Scope, stateScope), + ); + const exit = fiber.pollUnsafe(); + if (exit?._tag === "Failure" && !Cause.hasInterruptsOnly(exit.cause)) { + return yield* Effect.failCause(exit.cause); + } } }); diff --git a/test/actor.test.ts b/test/actor.test.ts index 9904193..b9af965 100644 --- a/test/actor.test.ts +++ b/test/actor.test.ts @@ -808,9 +808,7 @@ describe("ActorRef", () => { }) .final(TS.Done); - // Mirrors the gent AgentLoop pattern: send → yieldNow → waitFor(Running) - // The task fails immediately so Running→Idle can happen before - // waitFor subscribes. With the old get-then-subscribe waitFor, this hangs. + // The wait must own its subscription before the task can leave Running. interface LoopService { readonly run: () => Effect.Effect; } @@ -841,9 +839,11 @@ describe("ActorRef", () => { run: () => Effect.gen(function* () { const actor = yield* getActor; + const runningFiber = yield* Effect.forkDetach(actor.waitFor(TS.Running), { + startImmediately: true, + }); yield* actor.send(TE.Start); - yield* Effect.yieldNow; - yield* actor.waitFor(TS.Running); + yield* Fiber.join(runningFiber); yield* actor.waitFor((s) => s._tag !== "Running"); const final = yield* actor.snapshot; return final._tag; diff --git a/test/eager-effect-start.test.ts b/test/eager-effect-start.test.ts new file mode 100644 index 0000000..db0d5cd --- /dev/null +++ b/test/eager-effect-start.test.ts @@ -0,0 +1,232 @@ +// @effect-diagnostics strictEffectProvide:off - tests are entry points +import { Cause, Effect, Schema } from "effect"; +import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry"; +import { describe, expect, it } from "effect-bun-test"; + +import * as ActorAtom from "../src/atom.js"; +import { Event, Machine, State } from "../src/index.js"; + +const LifecycleState = State({ Idle: {}, Listening: {}, Done: {} }); +const LifecycleEvent = Event({ Start: {}, Stop: {} }); + +describe("actor Effect eager start", () => { + it.scopedLive("starts state Effect setup before state subscribers run", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, () => + Effect.sync(() => records.push("setup")).pipe(Effect.andThen(Effect.never)), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + const unsubscribe = actor.subscribe((state) => { + if (LifecycleState.$is("Listening")(state)) records.push("visible"); + }); + + yield* actor.call(LifecycleEvent.Start); + + expect(records).toEqual(["setup", "visible"]); + unsubscribe(); + yield* actor.stop; + }), + ); + + it.scopedLive("starts state Effects in registration order", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, () => + Effect.sync(() => records.push("first")).pipe(Effect.andThen(Effect.never)), + ) + .spawn(LifecycleState.Listening, () => + Effect.sync(() => records.push("second")).pipe(Effect.andThen(Effect.never)), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + const unsubscribe = actor.subscribe((state) => { + if (LifecycleState.$is("Listening")(state)) records.push("visible"); + }); + + yield* actor.call(LifecycleEvent.Start); + + expect(records).toEqual(["first", "second", "visible"]); + unsubscribe(); + yield* actor.stop; + }), + ); + + it.scopedLive("starts task setup before state subscribers run", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .task( + LifecycleState.Listening, + () => Effect.sync(() => records.push("task")).pipe(Effect.andThen(Effect.never)), + { onSuccess: () => LifecycleEvent.Stop }, + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + const unsubscribe = actor.subscribe((state) => { + if (LifecycleState.$is("Listening")(state)) records.push("visible"); + }); + + yield* actor.call(LifecycleEvent.Start); + + expect(records).toEqual(["task", "visible"]); + unsubscribe(); + yield* actor.stop; + }), + ); + + it.scopedLive("starts background Effect setup before actor start completes", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }).background(() => + Effect.sync(() => records.push("background")).pipe(Effect.andThen(Effect.never)), + ); + const actor = yield* Machine.spawn(machine); + + yield* actor.start; + + expect(records).toEqual(["background"]); + yield* actor.stop; + }), + ); + + it.scopedLive("interrupts an eagerly started state Effect on state exit", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .on(LifecycleState.Listening, LifecycleEvent.Stop, () => LifecycleState.Done) + .spawn(LifecycleState.Listening, () => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => Effect.sync(() => records.push("cleanup"))); + records.push("setup"); + return yield* Effect.never; + }), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + yield* actor.call(LifecycleEvent.Start); + yield* actor.call(LifecycleEvent.Stop); + + expect(records).toEqual(["setup", "cleanup"]); + yield* actor.stop; + }), + ); + + it.scopedLive("reports a synchronous state Effect defect as spawn", () => + Effect.gen(function* () { + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, () => Effect.die("spawn boom")); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + yield* actor.send(LifecycleEvent.Start); + const exit = yield* actor.awaitExit; + + expect(exit._tag).toBe("Defect"); + if (exit._tag === "Defect") { + expect(exit.phase).toBe("spawn"); + expect(Cause.hasInterruptsOnly(exit.cause)).toBe(false); + expect(Cause.pretty(exit.cause)).toContain("spawn boom"); + } + }), + ); + + it.scopedLive("settles a deferred ask from synchronous state Effect setup", () => + Effect.gen(function* () { + const ReplyState = State({ Idle: {}, Replying: {} }); + const ReplyEvent = Event({ Request: Event.reply({}, Schema.String) }); + const replyResults: Array = []; + const machine = Machine.make({ + state: ReplyState, + event: ReplyEvent, + initial: ReplyState.Idle, + }) + .on(ReplyState.Idle, ReplyEvent.Request, () => Machine.deferReply(ReplyState.Replying)) + .spawn(ReplyState.Replying, ({ self }) => + self.reply("ready").pipe( + Effect.tap((didReply) => Effect.sync(() => replyResults.push(didReply))), + Effect.andThen(Effect.never), + ), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + const reply = yield* Effect.race( + actor.ask(ReplyEvent.Request), + Effect.sleep("100 millis").pipe(Effect.as("timeout")), + ); + yield* Effect.yieldNow; + + expect(reply).toBe("ready"); + expect(replyResults).toEqual([true]); + yield* actor.stop; + }), + ); + + it.scopedLive("starts state Effect setup before Actor Atom subscribers run", () => + Effect.gen(function* () { + const records: Array = []; + const machine = Machine.make({ + state: LifecycleState, + event: LifecycleEvent, + initial: LifecycleState.Idle, + }) + .on(LifecycleState.Idle, LifecycleEvent.Start, () => LifecycleState.Listening) + .spawn(LifecycleState.Listening, () => + Effect.sync(() => records.push("setup")).pipe(Effect.andThen(Effect.never)), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + const registry = AtomRegistry.make(); + const stateAtom = ActorAtom.make(actor); + const unsubscribe = registry.subscribe( + stateAtom, + (state) => { + if (LifecycleState.$is("Listening")(state)) records.push("visible"); + }, + { immediate: true }, + ); + + yield* actor.call(LifecycleEvent.Start); + yield* Effect.yieldNow; + + expect(records).toEqual(["setup", "visible"]); + unsubscribe(); + registry.dispose(); + yield* actor.stop; + }), + ); +}); diff --git a/test/inspection.test.ts b/test/inspection.test.ts index dcb1012..b96332a 100644 --- a/test/inspection.test.ts +++ b/test/inspection.test.ts @@ -247,9 +247,9 @@ describe("Inspection", () => { }).spawn(TestState.Idle, () => Effect.die("boom")); const system = yield* ActorSystemService; - yield* system.spawn("test", machine); - yield* yieldFibers; + const spawnExit = yield* Effect.exit(system.spawn("test", machine)); + expect(spawnExit._tag).toBe("Failure"); const errorEvent = events.find((e) => e.type === "@machine.error"); expect(errorEvent).toBeDefined(); if (errorEvent?.type === "@machine.error") { diff --git a/test/supervision.test.ts b/test/supervision.test.ts index f4b629e..634137c 100644 --- a/test/supervision.test.ts +++ b/test/supervision.test.ts @@ -34,6 +34,31 @@ const machine = Machine.make({ state: S, event: E, initial: S.Idle }) // ============================================================================ describe("supervision: restart on defect", () => { + it.scopedLive("restarts after synchronous initial state Effect defect", () => + Effect.gen(function* () { + const restarted = yield* Deferred.make(); + let attempts = 0; + const initialEffectMachine = Machine.make({ state: S, event: E, initial: S.Idle }).spawn( + S.Idle, + () => + Effect.suspend(() => { + attempts += 1; + if (attempts === 1) return Effect.die("initial boom"); + return Deferred.succeed(restarted, void 0); + }), + ); + const actor = yield* Machine.spawn(initialEffectMachine, { + supervision: Supervision.restart({ maxRestarts: 1 }), + }); + + yield* actor.start; + yield* Deferred.await(restarted); + + expect(attempts).toBe(2); + yield* actor.stop; + }), + ); + it.scopedLive("restarts actor from initial state after transition defect", () => Effect.gen(function* () { const actor = yield* Machine.spawn(machine, {