diff --git a/.changeset/steady-actor-identity.md b/.changeset/steady-actor-identity.md new file mode 100644 index 0000000..3d044be --- /dev/null +++ b/.changeset/steady-actor-identity.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Run actor startup once across concurrent and repeated calls. Preserve terminal lifecycle when start is called after stop. Keep replacement actors registered when an earlier owner scope closes. diff --git a/AGENTS.md b/AGENTS.md index 3c3e543..6f2fc8d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -279,6 +279,8 @@ const count = yield* actor.ask(Event.GetCount); // number ## Gotchas +- `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. - Never `throw` in Effect.gen — use `yield* Effect.fail()` - `yield* Effect.yieldNow` after `send()` to let effects run diff --git a/src/actor.ts b/src/actor.ts index 04666cc..2e4f598 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -142,6 +142,8 @@ export interface ActorRef, }); - MutableHashMap.remove(actorsMap, id); } yield* actor.stop; }), diff --git a/test/actor-lifecycle.test.ts b/test/actor-lifecycle.test.ts index f9a5c04..92f895b 100644 --- a/test/actor-lifecycle.test.ts +++ b/test/actor-lifecycle.test.ts @@ -1,8 +1,15 @@ // @effect-diagnostics strictEffectProvide:off - tests are entry points -import { Deferred, Effect, Option } from "effect"; +import { Deferred, Effect, Exit, Fiber, Option, Scope } from "effect"; import { describe, expect, it } from "effect-bun-test"; -import { ActorSystemDefault, ActorSystemService, Event, Machine, State } from "../src/index.js"; +import { + ActorScope, + ActorSystemDefault, + ActorSystemService, + Event, + Machine, + State, +} from "../src/index.js"; const TestState = State({ Idle: {}, Done: {} }); const TestEvent = Event({ Finish: {} }); @@ -16,6 +23,138 @@ const machine = Machine.make({ .final(TestState.Done); describe("actor lifecycle observation", () => { + it.scopedLive("retains an interrupted first start until the actor is stopped", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + let recoveries = 0; + const actor = yield* Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Effect.gen(function* () { + recoveries++; + yield* Deferred.succeed(entered, undefined); + return yield* Effect.never; + }), + }, + }, + }); + const first = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + yield* Fiber.interrupt(first); + expect(Exit.hasInterrupts(yield* actor.start.pipe(Effect.exit))).toBe(true); + expect(recoveries).toBe(1); + yield* actor.stop; + expect(actor.client.getLifecycle()._tag).toBe("Stopped"); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scopedLive("keeps a replacement registered by a scope cleanup listener", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const owner = yield* Scope.make(); + const original = yield* system + .spawn("reentrant", machine) + .pipe(Effect.provideService(ActorScope, owner)); + const replacement = yield* Deferred.make(); + const unsubscribe = system.subscribe((event) => { + if (event._tag !== "ActorStopped" || event.id !== "reentrant") return; + unsubscribe(); + // @effect-diagnostics-next-line runEffectInsideEffect:off -- A synchronous system listener can start Effect work inline. + Effect.runFork( + system.stop("reentrant").pipe( + Effect.andThen(system.spawn("reentrant", machine)), + Effect.flatMap((actor) => Deferred.succeed(replacement, actor)), + Effect.orDie, + ), + ); + }); + yield* Effect.addFinalizer(() => Effect.sync(unsubscribe)); + yield* Scope.close(owner, Exit.void); + const actor = yield* Deferred.await(replacement); + expect(Object.is(system.actors.get("reentrant"), actor)).toBe(true); + yield* actor.send(TestEvent.Finish); + expect((yield* actor.awaitExit)._tag).toBe("Final"); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scopedLive("runs recovery once for concurrent and repeated starts", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + let recoveries = 0; + const actor = yield* Machine.spawn(machine, { + lifecycle: { + recovery: { + resolve: () => + Effect.gen(function* () { + recoveries++; + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + return Option.none(); + }), + }, + }, + }); + const first = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const second = yield* actor.start.pipe(Effect.forkScoped); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(first); + yield* Fiber.join(second); + yield* actor.start; + expect(recoveries).toBe(1); + expect(actor.client.getLifecycle()._tag).toBe("Active"); + yield* actor.stop; + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scopedLive("keeps terminal lifecycle when start is called again", () => + Effect.gen(function* () { + const unstarted = yield* Machine.spawn(machine); + yield* unstarted.stop; + const unstartedExit = yield* unstarted.awaitExit; + yield* unstarted.start; + expect(unstarted.client.getLifecycle()).toBe(unstartedExit); + + const stopped = yield* Machine.spawn(machine); + yield* stopped.start; + yield* stopped.stop; + const stoppedExit = yield* stopped.awaitExit; + yield* stopped.start; + expect(stopped.client.getLifecycle()).toBe(stoppedExit); + + const finished = yield* Machine.spawn(machine); + yield* finished.start; + yield* finished.send(TestEvent.Finish); + const finalExit = yield* finished.awaitExit; + yield* finished.start; + expect(finished.client.getLifecycle()).toBe(finalExit); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scopedLive("keeps a replacement actor when the previous owner scope closes", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const owner = yield* Scope.make(); + yield* system.spawn("reused", machine).pipe(Effect.provideService(ActorScope, owner)); + yield* system.stop("reused"); + const replacement = yield* system.spawn("reused", machine); + const registeredReplacement = system.actors.get("reused"); + expect(Object.is(registeredReplacement, replacement)).toBe(true); + const stoppedIds: string[] = []; + const unsubscribe = system.subscribe((event) => { + if (event._tag === "ActorStopped") stoppedIds.push(event.id); + }); + yield* Effect.addFinalizer(() => Effect.sync(unsubscribe)); + yield* Scope.close(owner, Exit.void); + expect(system.actors.get("reused")).toBe(registeredReplacement); + expect(stoppedIds).toEqual([]); + yield* replacement.send(TestEvent.Finish); + expect((yield* replacement.awaitExit)._tag).toBe("Final"); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + it.scopedLive("keeps lifecycle and latest transition after final exit", () => Effect.gen(function* () { const actor = yield* Machine.spawn(machine); @@ -25,7 +164,6 @@ describe("actor lifecycle observation", () => { yield* actor.send(TestEvent.Finish); yield* actor.awaitExit; - yield* Effect.yieldNow; expect(actor.client.getLifecycle()._tag).toBe("Final"); const latest = actor.client.getLatestTransition();