From 020964a70e4176e6752941f8592f630f2a89518c Mon Sep 17 00:00:00 2001 From: Cristian Date: Mon, 7 Sep 2026 08:24:26 +0000 Subject: [PATCH] feat: add lazy actors owned by parent state scopes --- .changeset/lazy-actor-host.md | 5 + AGENTS.md | 14 ++ README.md | 1 + SKILL.md | 6 + docs/actors.md | 27 +++ src/actor-host.ts | 173 +++++++++++++++ src/errors.ts | 12 ++ src/index.ts | 4 + test/actor-host.test.ts | 381 ++++++++++++++++++++++++++++++++++ 9 files changed, 623 insertions(+) create mode 100644 .changeset/lazy-actor-host.md create mode 100644 src/actor-host.ts create mode 100644 test/actor-host.test.ts diff --git a/.changeset/lazy-actor-host.md b/.changeset/lazy-actor-host.md new file mode 100644 index 0000000..79172a1 --- /dev/null +++ b/.changeset/lazy-actor-host.md @@ -0,0 +1,5 @@ +--- +"effect-machine": minor +--- + +Add ActorHost for lazy actors owned by a parent state scope. Consumers share startup without owning cancellation or shutdown. The factory captures service dependencies, uses the first matching request input, and releases the actor on state exit or host service shutdown. diff --git a/AGENTS.md b/AGENTS.md index 0534093..9bbad4f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -222,6 +222,20 @@ machine.spawn(State.Active, ({ self }) => - `self.spawn` returns `Effect` — use `Effect.orDie` in handlers - Every `ActorRef` has `actor.system` for child access: `actor.system.get("worker-1")` +## Lazy State-Owned Actors + +Use `ActorHost.make({ identity, spawn })` when consumers must request a child without owning its lifetime. + +- Construct the host in the service layer. `spawn` captures those services. +- Run `host.host(input)` in the parent state's `.spawn` handler. It registers a generation and waits for a consumer. +- Consumers call `host.acquire(input)`. Identity values match with `Object.is`. The first matching consumer supplies the factory input. +- Concurrent consumers share startup and its result. Cancelling one consumer does not cancel startup. ActorHost starts actors from either `Machine.spawn` or `system.spawn` before it returns them. +- The factory receives the host generation's Scope and ActorScope. State exit closes the child. Closing the host service also closes the current generation and fails pending consumers. +- A second active host fails with `ActorHostOccupiedError`. A closed generation fails pending acquisition with `ActorHostClosedError`. +- Keep session validity, authorization, and completion events in application services. ActorHost only owns actor creation and lifetime. +- A failed factory stays failed until the hosting scope closes. Handle expected factory errors in the spawn handler with a transition or event. `Effect.orDie` is appropriate only for invariant failures; it defects the parent and cannot support a later reentry. +- Once registered, `host` is interrupted when its generation closes. Consumers get `ActorHostClosedError` before actor cleanup starts. An acquisition belongs to the generation it observed; call `acquire` again after reentry. Pre-registration calls to a closed host service fail with `ActorHostClosedError`. + ## ActorRef API ```ts diff --git a/README.md b/README.md index 1ce627b..938f88d 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,7 @@ Effect Machine does not add an action queue or a second context system. | Actor-owned stream or resource | `.background` | | Autonomous machine sequence | `Machine.run` with `Effect.flatMap` | | Interactive multi-phase flow | Parent machine with child actors | +| Lazy child requested by consumers | `ActorHost` in a parent state scope | Effect requirements remain in `R`. A machine cannot start until the application provides every required service. Effectful transition handlers must have `never` in their error channel. Convert expected failures to states or events. diff --git a/SKILL.md b/SKILL.md index 376e3ac..3d2173b 100644 --- a/SKILL.md +++ b/SKILL.md @@ -365,3 +365,9 @@ Input machines require `input: (entityId) => Input`. Use `initializeState` only | `examples/react` | React Suspense and selector example | | `examples/solid` | Solid Suspense and selector example | | `docs` | User and migration guides | + +## Lazy actors owned by a parent state + +Create `ActorHost.make({ identity, spawn })` in a scoped service layer. Use `host.host(input)` in a state-scoped `.spawn` handler. Use `host.acquire(input)` from consumers. Identity uses `Object.is`. The first matching consumer supplies the spawn input. Concurrent consumers share startup. Consumer cancellation does not stop startup or the actor. The parent state scope owns the actor, and host service shutdown closes any active generation. Keep session validation in the application. + +ActorHost starts direct `Machine.spawn` results too. A registered `host` wait is interrupted on generation close; consumers receive `ActorHostClosedError` before actor cleanup. A consumer attached to an old generation must acquire again after reentry. Handle expected factory errors in the parent's spawn handler. Use `Effect.orDie` only for invariant failures, because it defects the parent. Both host errors are exported from the package root. diff --git a/docs/actors.md b/docs/actors.md index 32f6e36..24f9082 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -57,6 +57,33 @@ The child uses the same actor system. The child stops when the parent exits the This pattern replaces a root router that invokes one screen actor for each route. See [`actor-system.ts`](../examples/core/src/actor-system.ts). +## Lazy actors owned by a parent state + +Use `ActorHost` when consumers request a child but the parent state must own its lifetime. Construct the host in a scoped service layer. Keep session validation and authorization in that service. + +```ts +const menuHost = + yield * + ActorHost.make({ + identity: (input: { sessionId: string }) => input.sessionId, + spawn: (input) => Machine.spawn(menuMachine, { input }), + }); + +parent.spawn(State.Menu, ({ state }) => + menuHost.host({ sessionId: state.sessionId }).pipe(Effect.asVoid, Effect.orDie), +); + +const menu = yield * menuHost.acquire({ sessionId }); +``` + +`host` registers one generation in the current state scope. The first matching `acquire` supplies the factory input. Identity values match with `Object.is`. Concurrent consumers share startup and its result. Consumer cancellation does not cancel startup or stop the actor. ActorHost starts the factory result, so both `Machine.spawn` and `system.spawn` work. + +The factory uses the services captured when the host was made. Its Scope and ActorScope belong to the hosting generation. State exit closes the actor. Closing the host service also closes the current generation and ends consumers waiting for a future generation. + +Consumers receive `ActorHostClosedError` before actor cleanup starts. An acquisition belongs to the generation it observed. Acquire again after reentry to get the new actor. The registered `host` wait is interrupted when its generation closes. Calls made after the host service closes fail with `ActorHostClosedError`. An overlapping host fails with `ActorHostOccupiedError`, including while the previous actor is still being cleaned up. + +Factory failures stay in the typed error channel and remain shared until the hosting scope closes. Handle expected factory failures in the parent's spawn handler with a state transition or event. The example uses `Effect.orDie` because it treats the remaining host errors as wiring failures. A defect stops the parent; it cannot reenter to retry. Both host error classes are exported from the package root. + ## ActorRef selection - `send` queues an event and returns. diff --git a/src/actor-host.ts b/src/actor-host.ts new file mode 100644 index 0000000..9c8d688 --- /dev/null +++ b/src/actor-host.ts @@ -0,0 +1,173 @@ +import { Deferred, Effect, Exit, Option, Scope, Stream, SubscriptionRef } from "effect"; + +import type { ActorRef } from "./actor.js"; +import * as Machine from "./machine.js"; + +import { ActorHostClosedError, ActorHostOccupiedError } from "./errors.js"; + +export { ActorHostClosedError, ActorHostOccupiedError } from "./errors.js"; + +export interface ActorHost { + /** Register in the current scope. Wait for acquisition; interrupt if this generation closes. */ + readonly host: ( + input: Input, + ) => Effect.Effect< + ActorRef, + Failure | ActorHostClosedError | ActorHostOccupiedError, + Scope.Scope + >; + /** Wait for a matching generation. Caller cancellation does not cancel actor startup. */ + readonly acquire: ( + input: Input, + ) => Effect.Effect, Failure | ActorHostClosedError>; +} + +/** + * A lazy actor whose lifetime belongs to its host scope, not its consumers. + * + * Call `host` from a machine's state-scoped spawn handler. Call `acquire` from + * consumers. The first matching consumer supplies the spawn input. Identity + * values match with Object.is. The factory captures the services at make time; + * its Scope and ActorScope always belong to the hosting generation. The actor + * starts before publication, including factories that use Machine.spawn. + */ +export const make = Effect.fn("effect-machine.actorHost.make")(function* < + Input, + S extends { readonly _tag: string }, + E, + Output, + Failure, + R, +>(options: { + readonly identity: (input: Input) => unknown; + readonly spawn: (input: Input) => Effect.Effect, Failure, R>; +}) { + type Actor = ActorRef; + interface Entry { + readonly identity: unknown; + readonly scope: Scope.Closeable; + readonly requested: Deferred.Deferred; + readonly actor: Deferred.Deferred; + readonly closed: Deferred.Deferred; + } + const services = yield* Effect.context>(); + const current = yield* SubscriptionRef.make(Option.none()); + const closed = yield* Deferred.make(); + const failClosed = Effect.fail(ActorHostClosedError.make({})); + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + yield* Deferred.succeed(closed, undefined); + const entry = yield* SubscriptionRef.get(current); + if (Option.isSome(entry)) yield* Scope.close(entry.value.scope, Exit.void); + }), + ); + + const awaitActor = (entry: Entry): Effect.Effect => + Effect.raceFirst( + Deferred.await(entry.actor), + Deferred.await(entry.closed).pipe(Effect.andThen(failClosed)), + ).pipe( + Effect.flatMap((actor) => + Deferred.isDone(entry.closed).pipe( + Effect.flatMap((ended) => { + if (ended) return failClosed; + return Effect.succeed(actor); + }), + ), + ), + ); + + const host: ActorHost["host"] = Effect.fn( + "effect-machine.actorHost.host", + )((input) => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + if (yield* Deferred.isDone(closed)) return yield* failClosed; + const generation = yield* Scope.fork(yield* Scope.Scope); + const entry: Entry = { + identity: options.identity(input), + scope: generation, + requested: yield* Deferred.make(), + actor: yield* Deferred.make(), + closed: yield* Deferred.make(), + }; + yield* Scope.addFinalizer( + generation, + SubscriptionRef.update(current, (value) => { + if (Option.isSome(value) && value.value === entry) return Option.none(); + return value; + }), + ); + const scope = yield* Scope.fork(generation); + // Publish closure before startup cancellation; retire the entry after actor cleanup. + yield* Scope.addFinalizer( + generation, + Deferred.succeed(entry.closed, undefined).pipe( + Effect.andThen(Deferred.fail(entry.actor, ActorHostClosedError.make({}))), + ), + ); + const registered = yield* SubscriptionRef.modify(current, (value) => { + if ( + Option.isSome(value) || + Deferred.isDoneUnsafe(closed) || + Deferred.isDoneUnsafe(entry.closed) + ) { + return [false, value]; + } + return [true, Option.some(entry)]; + }); + if (!registered) { + const ended = (yield* Deferred.isDone(closed)) || (yield* Deferred.isDone(entry.closed)); + yield* Scope.close(generation, Exit.void); + if (ended) return yield* failClosed; + return yield* ActorHostOccupiedError.make({}); + } + yield* Effect.forkIn( + Deferred.complete( + entry.actor, + Deferred.await(entry.requested).pipe( + Effect.flatMap((requested) => + Machine.scoped(options.spawn(requested).pipe(Effect.tap((actor) => actor.start))), + ), + Scope.provide(scope), + Effect.provideContext(services), + ), + ), + scope, + ); + return yield* restore( + awaitActor(entry).pipe( + Effect.catchTag("ActorHostClosedError", (error) => + Deferred.isDone(entry.closed).pipe( + Effect.flatMap((ownGenerationClosed) => { + if (ownGenerationClosed) return Effect.interrupt; + return Effect.fail(error); + }), + ), + ), + ), + ); + }), + ), + ); + const acquire: ActorHost["acquire"] = Effect.fn( + "effect-machine.actorHost.acquire", + )((input) => + Effect.raceFirst( + Effect.gen(function* () { + const identity = options.identity(input); + const entry = yield* SubscriptionRef.changes(current).pipe( + Stream.filter(Option.isSome), + Stream.map((value) => value.value), + Stream.filter((value) => Object.is(value.identity, identity)), + Stream.runHead, + Effect.flatMap(Option.match({ onNone: () => failClosed, onSome: Effect.succeed })), + ); + yield* Deferred.succeed(entry.requested, input); + return yield* awaitActor(entry); + }), + Deferred.await(closed).pipe(Effect.andThen(failClosed)), + ), + ); + return { host, acquire } satisfies ActorHost; +}); diff --git a/src/errors.ts b/src/errors.ts index 81be7ca..7caf323 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -55,3 +55,15 @@ export class VersionConflictError extends Schema.TaggedError()( + "ActorHostClosedError", + {}, +) {} + +/** A host can own only one state generation at a time. */ +export class ActorHostOccupiedError extends Schema.TaggedError()( + "ActorHostOccupiedError", + {}, +) {} diff --git a/src/index.ts b/src/index.ts index cbe7350..78e72ae 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,8 +1,12 @@ // Machine namespace (Effect-style) export * as Machine from "./machine.js"; +export * as ActorHost from "./actor-host.js"; + // Errors export { + ActorHostClosedError, + ActorHostOccupiedError, ActorStoppedError, AssertionError, DuplicateActorError, diff --git a/test/actor-host.test.ts b/test/actor-host.test.ts new file mode 100644 index 0000000..2719fd9 --- /dev/null +++ b/test/actor-host.test.ts @@ -0,0 +1,381 @@ +// @effect-diagnostics strictEffectProvide:off - tests are entry points +import { + Cause, + Context, + Deferred, + Effect, + Exit, + Fiber, + Option, + Schema, + Scope, + SubscriptionRef, +} from "effect"; +import { describe, expect, it, yieldFibers } from "effect-bun-test"; + +import { + ActorHost, + ActorSystemDefault, + ActorSystemService, + Event, + Machine, + State, +} from "../src/index.js"; + +const ChildState = State({ Ready: { value: Schema.String }, Done: {} }); +const ChildEvent = Event({ Finish: {} }); +const child = Machine.make({ + state: ChildState, + event: ChildEvent, + initial: (input: { value: string }) => ChildState.Ready(input), +}) + .on(ChildState.Ready, ChildEvent.Finish, () => ChildState.Done) + .final(ChildState.Done); + +class Label extends Context.Service()( + "effect-machine/test/actor-host.test/Label", +) {} + +const ownerScope = Effect.acquireRelease(Scope.make(), (scope) => Scope.close(scope, Exit.void)); + +describe("ActorHost", () => { + it.scoped( + "is lazy, uses request input and captured services, and survives consumer scope close", + () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const host = yield* ActorHost.make({ + identity: (input: { session: object; value: string }) => input.session, + spawn: (input) => + Effect.gen(function* () { + const label = yield* Label; + return yield* system.spawn("child", child, { + input: { value: `${label.value}:${input.value}` }, + }); + }), + }).pipe(Effect.provideService(Label, { value: "root" })); + const session = {}; + const owner = yield* ownerScope; + const hosted = yield* host + .host({ session, value: "old" }) + .pipe(Scope.provide(owner), Effect.forkScoped); + yield* yieldFibers; + expect(Option.isNone(yield* system.get("child"))).toBe(true); + const [first, second] = yield* Effect.all( + [ + Effect.scoped(host.acquire({ session, value: "current" })), + host.acquire({ session, value: "current" }), + ], + { concurrency: "unbounded" }, + ).pipe(Effect.provideService(Label, { value: "consumer" })); + expect(first).toBe(second); + expect(yield* Fiber.join(hosted)).toBe(first); + expect(yield* first.snapshot).toEqual(ChildState.Ready({ value: "root:current" })); + expect(Option.isSome(yield* system.get("child"))).toBe(true); + yield* Scope.close(owner, Exit.void); + expect(Option.isNone(yield* system.get("child"))).toBe(true); + expect((yield* first.call(ChildEvent.Finish)).transitioned).toBe(false); + const nextOwner = yield* ownerScope; + const nextHosted = yield* host + .host({ session, value: "unused" }) + .pipe(Scope.provide(nextOwner), Effect.forkScoped); + const next = yield* host.acquire({ session, value: "next" }); + expect(next).not.toBe(first); + expect(yield* Fiber.join(nextHosted)).toBe(next); + expect(yield* next.snapshot).toEqual(ChildState.Ready({ value: "root:next" })); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped("keeps actual recovery running when an acquiring consumer cancels", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => + system.spawn("recovering", child, { + input: { value }, + lifecycle: { + recovery: { + resolve: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as(Option.none()), + ), + }, + }, + }), + }); + const owner = yield* ownerScope; + const hosted = yield* host.host("session").pipe(Scope.provide(owner), Effect.forkScoped); + const first = yield* host.acquire("session").pipe(Effect.forkScoped); + yield* Deferred.await(started); + yield* Fiber.interrupt(first); + const second = yield* host.acquire("session").pipe(Effect.forkScoped); + yield* Deferred.succeed(release, undefined); + const actor = yield* Fiber.join(second); + expect(yield* Fiber.join(hosted)).toBe(actor); + expect(yield* actor.snapshot).toEqual(ChildState.Ready({ value: "session" })); + yield* Scope.close(owner, Exit.void); + expect(Option.isNone(yield* system.get("recovering"))).toBe(true); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped("closes pending recovery and all waiting acquisitions with its owner", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const started = yield* Deferred.make(); + const released = yield* Deferred.make(); + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => + system.spawn("pending", child, { + input: { value }, + lifecycle: { + recovery: { + resolve: () => + Effect.acquireUseRelease( + Deferred.succeed(started, undefined), + () => Effect.never, + () => Deferred.succeed(released, undefined), + ), + }, + }, + }), + }); + const owner = yield* ownerScope; + const hosted = yield* host + .host("session") + .pipe(Scope.provide(owner), Effect.exit, Effect.forkScoped); + const acquisition = yield* host.acquire("session").pipe(Effect.flip, Effect.forkScoped); + yield* Deferred.await(started); + yield* Scope.close(owner, Exit.void); + expect(yield* Deferred.isDone(released)).toBe(true); + const hostExit = yield* Fiber.join(hosted); + expect(Exit.isFailure(hostExit)).toBe(true); + if (Exit.isFailure(hostExit)) expect(Cause.hasInterruptsOnly(hostExit.cause)).toBe(true); + expect((yield* Fiber.join(acquisition))._tag).toBe("ActorHostClosedError"); + expect(Option.isNone(yield* system.get("pending"))).toBe(true); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped("rejects overlapping hosts and closes unmatched consumers with the service", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const serviceScope = yield* ownerScope; + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => system.spawn("child", child, { input: { value } }), + }).pipe(Scope.provide(serviceScope)); + const owner = yield* ownerScope; + const hosted = yield* host + .host("first") + .pipe(Scope.provide(owner), Effect.exit, Effect.forkScoped); + yield* yieldFibers; + const otherOwner = yield* ownerScope; + const duplicate = yield* host.host("second").pipe(Scope.provide(otherOwner), Effect.flip); + expect(duplicate._tag).toBe("ActorHostOccupiedError"); + const unmatched = yield* host.acquire("other").pipe(Effect.flip, Effect.forkScoped); + yield* Scope.close(serviceScope, Exit.void); + expect((yield* Fiber.join(unmatched))._tag).toBe("ActorHostClosedError"); + const hostExit = yield* Fiber.join(hosted); + expect(Exit.isFailure(hostExit)).toBe(true); + if (Exit.isFailure(hostExit)) expect(Cause.hasInterruptsOnly(hostExit.cause)).toBe(true); + expect((yield* host.acquire("first").pipe(Effect.flip))._tag).toBe("ActorHostClosedError"); + expect((yield* host.host("first").pipe(Scope.provide(otherOwner), Effect.flip))._tag).toBe( + "ActorHostClosedError", + ); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped("replaces the child on native parent state reentry and stops it on parent close", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => system.spawn("state-child", child, { input: { value } }), + }); + const ParentState = State({ Inactive: {}, Active: {} }); + const ParentEvent = Event({ Enter: {}, Leave: {}, Restart: {} }); + const parentMachine = Machine.make({ + state: ParentState, + event: ParentEvent, + initial: ParentState.Inactive, + }) + .on(ParentState.Inactive, ParentEvent.Enter, () => ParentState.Active) + .on(ParentState.Active, ParentEvent.Leave, () => ParentState.Inactive) + .reenter(ParentState.Active, ParentEvent.Restart, () => ParentState.Active) + .spawn(ParentState.Active, () => host.host("session").pipe(Effect.orDie, Effect.asVoid)); + const parent = yield* Machine.scoped(system.spawn("parent", parentMachine)); + yield* parent.call(ParentEvent.Enter); + yield* parent.call(ParentEvent.Leave); + expect(Option.isNone(yield* system.get("state-child"))).toBe(true); + const pending = yield* host.acquire("session").pipe(Effect.forkScoped); + yield* parent.call(ParentEvent.Enter); + const first = yield* Fiber.join(pending); + yield* parent.call(ParentEvent.Restart); + const second = yield* host.acquire("session"); + expect(second).not.toBe(first); + expect((yield* first.call(ChildEvent.Finish)).transitioned).toBe(false); + yield* parent.call(ParentEvent.Leave); + expect(Option.isNone(yield* system.get("state-child"))).toBe(true); + yield* parent.call(ParentEvent.Enter); + const third = yield* host.acquire("session"); + expect(third).not.toBe(second); + yield* parent.stop; + expect(Option.isNone(yield* system.get("state-child"))).toBe(true); + expect((yield* third.call(ChildEvent.Finish)).transitioned).toBe(false); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped( + "shares typed startup failure and stops a live actor when its host service closes", + () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const existing = yield* Machine.scoped( + system.spawn("shared", child, { input: { value: "existing" } }), + ); + const serviceScope = yield* ownerScope; + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => system.spawn("shared", child, { input: { value } }), + }).pipe(Scope.provide(serviceScope)); + const owner = yield* ownerScope; + const hosted = yield* host + .host("session") + .pipe(Scope.provide(owner), Effect.flip, Effect.forkScoped); + const failure = yield* host.acquire("session").pipe(Effect.flip); + expect(failure._tag).toBe("DuplicateActorError"); + expect(yield* Fiber.join(hosted)).toBe(failure); + expect(yield* host.acquire("session").pipe(Effect.flip)).toBe(failure); + yield* Scope.close(owner, Exit.void); + yield* existing.stop; + const nextOwner = yield* ownerScope; + const nextHosted = yield* host + .host("session") + .pipe(Scope.provide(nextOwner), Effect.forkScoped); + const actor = yield* host.acquire("session"); + expect(yield* Fiber.join(nextHosted)).toBe(actor); + yield* Scope.close(serviceScope, Exit.void); + expect(Option.isNone(yield* system.get("shared"))).toBe(true); + expect((yield* actor.call(ChildEvent.Finish)).transitioned).toBe(false); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + it.scoped("starts actors returned by Machine.spawn before publishing them", () => + Effect.gen(function* () { + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => Machine.spawn(child, { input: { value } }), + }); + const owner = yield* ownerScope; + const hosted = yield* host.host("direct").pipe(Scope.provide(owner), Effect.forkScoped); + const actor = yield* host.acquire("direct"); + expect((yield* SubscriptionRef.get(actor.lifecycle))._tag).toBe("Active"); + expect(yield* Fiber.join(hosted)).toBe(actor); + expect((yield* actor.call(ChildEvent.Finish)).transitioned).toBe(true); + }), + ); + it.scoped("ends a pending native state acquisition with a typed error and permits reentry", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => + system.spawn("native-recovery", child, { + input: { value }, + lifecycle: { + recovery: { + resolve: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as(Option.none()), + ), + }, + }, + }), + }); + const ParentState = State({ Inactive: {}, Active: {} }); + const ParentEvent = Event({ Enter: {}, Leave: {} }); + const parentMachine = Machine.make({ + state: ParentState, + event: ParentEvent, + initial: ParentState.Inactive, + }) + .on(ParentState.Inactive, ParentEvent.Enter, () => ParentState.Active) + .on(ParentState.Active, ParentEvent.Leave, () => ParentState.Inactive) + .spawn(ParentState.Active, () => host.host("session").pipe(Effect.orDie, Effect.asVoid)); + const parent = yield* Machine.scoped(system.spawn("native-parent", parentMachine)); + yield* parent.call(ParentEvent.Enter); + const pending = yield* host.acquire("session").pipe(Effect.flip, Effect.forkScoped); + yield* Deferred.await(started); + yield* parent.call(ParentEvent.Leave); + expect((yield* Fiber.join(pending))._tag).toBe("ActorHostClosedError"); + expect(Option.isNone(yield* system.get("native-recovery"))).toBe(true); + expect((yield* parent.call(ParentEvent.Enter)).transitioned).toBe(true); + yield* Deferred.succeed(release, undefined); + const actor = yield* host.acquire("session"); + expect((yield* actor.snapshot)._tag).toBe("Ready"); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + + it.scoped("fails acquisitions during actor cleanup and rejects an overlapping generation", () => + Effect.gen(function* () { + const system = yield* ActorSystemService; + const stopping = yield* Deferred.make(); + const release = yield* Deferred.make(); + const cleanupChild = Machine.make({ + state: ChildState, + event: ChildEvent, + initial: (input: { value: string }) => ChildState.Ready(input), + }).background(() => + Effect.addFinalizer(() => + Deferred.succeed(stopping, undefined).pipe(Effect.andThen(Deferred.await(release))), + ), + ); + const host = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => system.spawn("cleanup", cleanupChild, { input: { value } }), + }); + const owner = yield* ownerScope; + yield* host.host("session").pipe(Scope.provide(owner), Effect.forkScoped); + yield* host.acquire("session"); + const closing = yield* Scope.close(owner, Exit.void).pipe(Effect.forkScoped); + yield* Deferred.await(stopping); + expect((yield* host.acquire("session").pipe(Effect.flip))._tag).toBe("ActorHostClosedError"); + const nextOwner = yield* ownerScope; + expect((yield* host.host("session").pipe(Scope.provide(nextOwner), Effect.flip))._tag).toBe( + "ActorHostOccupiedError", + ); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(closing); + yield* host.host("session").pipe(Scope.provide(nextOwner), Effect.forkScoped); + const next = yield* host.acquire("session"); + expect((yield* next.snapshot)._tag).toBe("Ready"); + }).pipe(Effect.provide(ActorSystemDefault)), + ); + it.scoped("preserves a factory error from another closed host", () => + Effect.gen(function* () { + const innerScope = yield* ownerScope; + const inner = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (value) => Machine.spawn(child, { input: { value } }), + }).pipe(Scope.provide(innerScope)); + yield* Scope.close(innerScope, Exit.void); + const outer = yield* ActorHost.make({ + identity: (input: string) => input, + spawn: (input) => inner.acquire(input), + }); + const owner = yield* ownerScope; + const hosted = yield* outer + .host("session") + .pipe(Scope.provide(owner), Effect.flip, Effect.forkScoped); + const error = yield* outer.acquire("session").pipe(Effect.flip); + expect(error._tag).toBe("ActorHostClosedError"); + expect(yield* Fiber.join(hosted)).toBe(error); + }), + ); +});