From 8e14f413e0f82e09d727fb736d5242c1162e51b8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 09:40:05 +0000 Subject: [PATCH] Deliver early atom machine commands and add bridge helper types Writes to MachineAtom and ChildMachineAtom send/stop made while the machine is still starting are now held and delivered in write order instead of failing with NotReadyError, which is removed. Adds MachineAtom.Any/State/Event and ChildMachineAtom.Any/State helpers. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .changeset/atom-send-buffering.md | 15 ++ .../effect-machine-react/src/MachineAtom.ts | 8 +- .../src/internal/react.ts | 6 +- .../test/MachineAtom.test.tsx | 42 ++- .../effect-machine/docs/effect-atom-react.md | 29 ++ .../src/internal/machine/atom.ts | 193 +++++++++----- .../src/unstable/reactivity/AtomMachine.ts | 110 ++++++-- .../reactivity/AtomMachineCommands.test.ts | 247 ++++++++++++++++++ .../unstable/reactivity/AtomMachine.tst.ts | 38 +++ 9 files changed, 585 insertions(+), 103 deletions(-) create mode 100644 .changeset/atom-send-buffering.md create mode 100644 packages/effect-machine/test/unstable/reactivity/AtomMachineCommands.test.ts diff --git a/.changeset/atom-send-buffering.md b/.changeset/atom-send-buffering.md new file mode 100644 index 00000000..bec64984 --- /dev/null +++ b/.changeset/atom-send-buffering.md @@ -0,0 +1,15 @@ +--- +"@typeonce/effect-machine": minor +--- + +Deliver `send` and `stop` writes made while an atom machine is still starting. Writes to `MachineAtom.send`, `MachineAtom.stop`, `ChildMachineAtom.send` and `ChildMachineAtom.stop` before readiness are now held and delivered in write order once the machine is ready, instead of being dropped. The write result stays waiting until delivery and fails with the startup error if startup fails. Writes after the machine stopped still fail with `StoppedError`, and child writes still fail with `ChildNotActiveError` when the owning state is inactive. + +`NotReadyError` is removed from `AtomMachine` and from the `send` and `stop` failure types. Remove any handling of it. + +Add `AtomMachine.MachineAtom.Any` and `AtomMachine.ChildMachineAtom.Any` for code that accepts any bridge, including bridges for machines without events, and the `MachineAtom.State`, `MachineAtom.Event` and `ChildMachineAtom.State` extractors: + +```ts +type Bridge = AtomMachine.MachineAtom.Any +type State = AtomMachine.MachineAtom.State +type Event = AtomMachine.MachineAtom.Event +``` diff --git a/packages/effect-machine-react/src/MachineAtom.ts b/packages/effect-machine-react/src/MachineAtom.ts index ff319799..79d24264 100644 --- a/packages/effect-machine-react/src/MachineAtom.ts +++ b/packages/effect-machine-react/src/MachineAtom.ts @@ -10,8 +10,6 @@ import type { AtomMachine } from "@typeonce/effect-machine/reactivity" import type * as React from "react" import * as internal from "./internal/react.js" -type AnyMachineAtom = AtomMachine.MachineAtom - /** * Creates one bridge per React owner and mounts its reference without subscribing * the owner to state changes. The factory is startup-only. Change the owner's @@ -21,7 +19,7 @@ type AnyMachineAtom = AtomMachine.MachineAtom * @category hooks * @since 0.29.0 */ -export const useMachineAtom: (create: () => A) => A = internal.useMachineAtom +export const useMachineAtom: (create: () => A) => A = internal.useMachineAtom /** Props for a typed render of one active state path. * @category models @@ -66,7 +64,7 @@ export const MachineState: < * @category models * @since 0.32.0 */ -export interface MachineContext { +export interface MachineContext { readonly Provider: React.ComponentType< & { readonly children?: React.ReactNode } & (Args extends [] ? { readonly input?: never } : { readonly input: Args[0] }) @@ -82,6 +80,6 @@ export interface MachineContext( +export const createMachineContext: ( create: (...args: Args) => A ) => MachineContext = internal.createMachineContext diff --git a/packages/effect-machine-react/src/internal/react.ts b/packages/effect-machine-react/src/internal/react.ts index 47d6e8e0..2a229cea 100644 --- a/packages/effect-machine-react/src/internal/react.ts +++ b/packages/effect-machine-react/src/internal/react.ts @@ -5,9 +5,7 @@ import { Option } from "effect" import * as React from "react" import type { MachineContext, MachineStateProps } from "../MachineAtom.js" -type AnyMachineAtom = AtomMachine.MachineAtom - -export const useMachineAtom = (create: () => A): A => { +export const useMachineAtom = (create: () => A): A => { const [machine] = React.useState(create) useAtomMount(machine.ref) return machine @@ -32,7 +30,7 @@ export const MachineState = < return Option.isSome(snapshot) ? props.children(snapshot.value) : props.inactive ?? null } -export const createMachineContext = ( +export const createMachineContext = ( create: (...args: Args) => A ): MachineContext => { const Context = React.createContext(undefined) diff --git a/packages/effect-machine-react/test/MachineAtom.test.tsx b/packages/effect-machine-react/test/MachineAtom.test.tsx index 45477831..f11f4bc0 100644 --- a/packages/effect-machine-react/test/MachineAtom.test.tsx +++ b/packages/effect-machine-react/test/MachineAtom.test.tsx @@ -1,7 +1,7 @@ // @vitest-environment jsdom -import { RegistryContext, useAtomSuspense } from "@effect/atom-react" +import { RegistryContext, useAtomSet, useAtomSuspense } from "@effect/atom-react" import { act, cleanup, render, screen, waitFor } from "@testing-library/react" -import { Effect, Equal, Option, Schema } from "effect" +import { Effect, Equal, Layer, Option, Schema } from "effect" import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity" import * as React from "react" import { renderToString } from "react-dom/server" @@ -243,6 +243,44 @@ describe("useMachineAtom", () => { }) registry.dispose() }) + it("delivers events sent on first mount while a bound runtime is still building", async () => { + const machine = trackedMachine(() => {}) + const runtime = Atom.runtime(Layer.effectDiscard(Effect.sleep("20 millis"))) + const registry = AtomRegistry.make() + function Reader({ owned }: { readonly owned: AtomMachine.MachineAtom.Any }) { + const active = useAtomSuspense(AtomMachine.matches(owned, "Active")).value + return {String(active)} + } + function Owner() { + const owned = useMachineAtom(() => AtomMachine.bind(runtime).make(machine, 3)) + const send = useAtomSet(owned.send) + React.useEffect(() => { + send(new Increment({})) + send(new Increment({})) + }, [send]) + const value = AtomMachine.select(owned, "Active") + return ( + + + + + ) + } + function Value({ value }: { + readonly value: Atom.Atom, unknown>> + }) { + const active = useAtomSuspense(value).value + return {Option.getOrThrow(active).value} + } + render( + + + + ) + await waitFor(() => assert.strictEqual(screen.getByTestId("value").textContent, "5")) + assert.strictEqual(screen.getByTestId("active").textContent, "true") + registry.dispose() + }) it("does not start the machine during server rendering", () => { let starts = 0 const machine = trackedMachine(() => { diff --git a/packages/effect-machine/docs/effect-atom-react.md b/packages/effect-machine/docs/effect-atom-react.md index e89960f3..f407c39a 100644 --- a/packages/effect-machine/docs/effect-atom-react.md +++ b/packages/effect-machine/docs/effect-atom-react.md @@ -159,6 +159,14 @@ function SubmitButton() { `useAtomSet` mounts the writable atom and does not subscribe the component to its value. +Writes made while the machine is still starting, for example while a bound +`Atom.runtime` layer is building after hydration, are held and delivered in +write order once the machine is ready. The write result stays waiting until +the event is delivered and fails with the startup error if startup fails. +Writes after the machine stopped or completed fail with `StoppedError`. The +same applies to `stop` and to child `send` and `stop`; a child command fails +with `ChildNotActiveError` when the state that owns the child is not active. + ## Query concrete event acceptance `AtomMachine.can` turns one concrete event input into a reusable machine @@ -328,6 +336,27 @@ export const plantAtoms = AtomMachine.familyChild(CentralMachineAtom, { }) ``` +## Generic helpers + +Use `AtomMachine.MachineAtom.Any` and `AtomMachine.ChildMachineAtom.Any` to +accept any bridge, including bridges for machines without events. Recover the +exact types with the namespace extractors: + +```ts +function StateView(props: { + readonly machine: A + readonly render: (state: AtomMachine.MachineAtom.State) => React.ReactNode +}) { + const state = useAtomSuspense(props.machine.result).value + return props.render(state) +} + +type AuthEvent = AtomMachine.MachineAtom.Event +type EditorState = AtomMachine.ChildMachineAtom.State< + AtomMachine.ChildOf +> +``` + ## Registry and rendering semantics A `MachineAtom` identifies one machine per `AtomRegistry`. Passing the same diff --git a/packages/effect-machine/src/internal/machine/atom.ts b/packages/effect-machine/src/internal/machine/atom.ts index d5a335da..d88253e7 100644 --- a/packages/effect-machine/src/internal/machine/atom.ts +++ b/packages/effect-machine/src/internal/machine/atom.ts @@ -21,8 +21,6 @@ import type { EnsureExecutable } from "./readiness.js" import type { ExcludeCompatibleRuntime as ExcludeCompatibleMachineRuntime, IsAny } from "./requirements.js" import * as Topology from "./topology.js" -export class NotReadyError extends Data.TaggedError("NotReadyError") {} - export class ChildNotActiveError extends Data.TaggedError("ChildNotActiveError")<{ readonly id: string }> {} @@ -213,6 +211,100 @@ export const childEmissions = = Atom.readable((get) => get.registry).pipe( + Atom.keepAlive +) + +/** + * Ordered delivery of held commands for one bridge in one registry. + * + * Commands written while the machine ref is still `Initial` are held here and + * delivered in write order once it settles. While held commands remain, later + * commands join the same queue so they cannot overtake earlier writes. + */ +interface CommandQueue { + readonly pending: Array> + draining: boolean +} + +type CommandQueues = WeakMap + +const drainCommands = (queue: CommandQueue): Effect.Effect => + Effect.suspend(() => { + const next = queue.pending.shift() + if (next === undefined) { + queue.draining = false + return Effect.void + } + return Effect.andThen(next, drainCommands(queue)) + }) + +const getCommandQueue = (queues: CommandQueues, registry: AtomRegistry.AtomRegistry): CommandQueue => { + let queue = queues.get(registry) + if (queue === undefined) { + queue = { pending: [], draining: false } + queues.set(registry, queue) + } + return queue +} + +/** + * Builds a command atom over a bridge ref. + * + * A write runs the command at once when the ref is ready and no earlier + * command is held. Otherwise the command is held until the ref settles, the + * write result stays waiting, and it settles with the command exit or with the + * startup failure. + */ +const makeCommandAtom = ( + ref: Atom.Atom>, + queues: CommandQueues, + command: (ref: Ref, value: W) => Effect.Effect +): Atom.Writable, W> => + Atom.writable( + (get): AsyncResult.AsyncResult => { + const current = AsyncResult.map(get(ref), () => undefined) + const queue = queues.get(get.registry) + if (queue === undefined || !queue.draining) { + return current + } + return Option.getOrElse( + Option.map(get.self>(), AsyncResult.waiting), + () => AsyncResult.initial(true) + ) + }, + (ctx, value) => { + const current = ctx.get(ref) + const registry = ctx.get(registryAtom) + const queue = getCommandQueue(queues, registry) + if (AsyncResult.isInitial(current) || queue.draining) { + ctx.setSelf(AsyncResult.initial(true)) + queue.pending.push( + AtomRegistry.getResult(registry, ref).pipe( + Effect.flatMap((ready) => command(ready, value)), + Effect.exit, + Effect.map((exit) => { + const result = AsyncResult.fromExit(exit) + ctx.setSelf(queue.pending.length === 0 ? result : AsyncResult.waiting(result)) + }) + ) + ) + if (!queue.draining) { + queue.draining = true + Effect.runFork(drainCommands(queue)) + } + } else if (AsyncResult.isFailure(current)) { + ctx.setSelf(AsyncResult.map(current, () => undefined)) + } else { + Effect.runCallback(command(current.value, value), { + onExit: (exit) => ctx.setSelf(AsyncResult.fromExit(exit)) + }) + } + } + ) + const makeRuntimeResultAtom = ( snapshot: Atom.Atom, StartError>> ): Atom.Atom> => @@ -338,44 +430,23 @@ const makeChildFromRefAtom = , - Machine.ChildMachine.Event - >( - (get) => AsyncResult.map(get(ref), () => undefined), - (ctx, event) => { - const result = ctx.get(ref) - if (AsyncResult.isInitial(result)) { - ctx.setSelf(AsyncResult.fail(new NotReadyError())) - } else if (AsyncResult.isFailure(result)) { - ctx.setSelf(AsyncResult.map(result, () => undefined)) - } else if (Option.isNone(result.value)) { - ctx.setSelf(AsyncResult.fail(new ChildNotActiveError({ id: descriptor.id }))) - } else { - Effect.runCallback(result.value.value.send(event as never), { - onExit: (exit) => ctx.setSelf(AsyncResult.fromExit(exit)) - }) - } - } + const commands: CommandQueues = new WeakMap() + const activeChild = (current: Option.Option): Effect.Effect => + Option.isSome(current) + ? Effect.succeed(current.value) + : Effect.fail(new ChildNotActiveError({ id: descriptor.id })) + + const send: ChildMachineAtom["send"] = makeCommandAtom( + ref, + commands, + (current, event: Machine.ChildMachine.Event) => + Effect.flatMap(activeChild(current), (child) => child.send(event as never)) ) - const stop = Atom.writable< - AsyncResult.AsyncResult, - void - >( - (get) => AsyncResult.map(get(ref), () => undefined), - (ctx) => { - const result = ctx.get(ref) - if (AsyncResult.isInitial(result)) { - ctx.setSelf(AsyncResult.fail(new NotReadyError())) - } else if (AsyncResult.isFailure(result)) { - ctx.setSelf(AsyncResult.map(result, () => undefined)) - } else if (Option.isNone(result.value)) { - ctx.setSelf(AsyncResult.fail(new ChildNotActiveError({ id: descriptor.id }))) - } else { - Effect.runCallback(result.value.value.stop) - } - } + const stop: ChildMachineAtom["stop"] = makeCommandAtom( + ref, + commands, + (current, _: void) => Effect.flatMap(activeChild(current), (child) => child.stop) ) const child = makeChildSelector(ref as any) @@ -453,40 +524,18 @@ const makeFromRefAtom = ( }) }) - const send = Atom.writable< - AsyncResult.AsyncResult, - Event - >( - (get) => AsyncResult.map(get(ref), () => undefined), - (ctx, event: Event) => { - const result = ctx.get(ref) - if (AsyncResult.isInitial(result)) { - ctx.setSelf(AsyncResult.fail(new NotReadyError())) - } else if (AsyncResult.isFailure(result)) { - ctx.setSelf(AsyncResult.map(result, () => undefined)) - } else { - Effect.runCallback(result.value.send(event), { - onExit: (exit) => - ctx.setSelf( - AsyncResult.fromExit(exit) - ) - }) - } - } + const commands: CommandQueues = new WeakMap() + + const send: MachineAtom["send"] = makeCommandAtom( + ref, + commands, + (handle, event: Event) => handle.send(event) ) - const stop = Atom.writable, void>( - (get) => AsyncResult.map(get(ref), () => undefined), - (ctx) => { - const result = ctx.get(ref) - if (AsyncResult.isInitial(result)) { - ctx.setSelf(AsyncResult.fail(new NotReadyError())) - } else if (AsyncResult.isFailure(result)) { - ctx.setSelf(AsyncResult.map(result, () => undefined)) - } else { - Effect.runCallback(result.value.stop) - } - } + const stop: MachineAtom["stop"] = makeCommandAtom( + ref, + commands, + (handle, _: void) => handle.stop ) const optionalRef = Atom.mapResult(ref, Option.some) @@ -882,7 +931,7 @@ const resumeWithRuntime = ( return makeFromRefAtom(ref as any, machine) } -type FamilyBridge = MachineAtom | ChildMachineAtom +type FamilyBridge = MachineAtom.Any | ChildMachineAtom.Any type FamilyOptions = { readonly atoms: Readonly Atom.Atom>> @@ -937,7 +986,7 @@ export const family = ( ) export const familyChild = ( - parent: MachineAtom | ChildMachineAtom, + parent: FamilyBridge, options: FamilyOptions & { readonly child: (key: any) => Machine.ChildMachine.Any } ): Readonly Atom.Atom>> => makeFamily( diff --git a/packages/effect-machine/src/unstable/reactivity/AtomMachine.ts b/packages/effect-machine/src/unstable/reactivity/AtomMachine.ts index 6ff8a877..1ee25d8e 100644 --- a/packages/effect-machine/src/unstable/reactivity/AtomMachine.ts +++ b/packages/effect-machine/src/unstable/reactivity/AtomMachine.ts @@ -11,7 +11,7 @@ import type * as Schema from "effect/Schema" import type * as Scope from "effect/Scope" import type * as Stream from "effect/Stream" import * as internal from "../../internal/machine/atom.js" -import type { ChildNotActiveError, NotReadyError } from "../../internal/machine/atom.js" +import type { ChildNotActiveError } from "../../internal/machine/atom.js" import type { EnsureExecutable } from "../../internal/machine/readiness.js" import type { ExcludeCompatibleRuntime as ExcludeCompatibleMachineRuntime, @@ -19,14 +19,6 @@ import type { } from "../../internal/machine/requirements.js" import type * as Machine from "../../Machine.js" -/** - * Error returned when a machine command is issued before startup completes. - * - * @category errors - * @since 0.4.0 - */ -export { NotReadyError } from "../../internal/machine/atom.js" - /** * Error returned when a command targets a child machine that is not active. * @@ -114,23 +106,34 @@ export interface MachineAtom> /** - * Writable atom that sends events to the machine. Writes before startup - * completes fail with `NotReadyError`. + * Writable atom that sends events to the machine. + * + * **Details** + * + * Events written while the machine is still starting are held and delivered + * in write order once startup succeeds. The write result is waiting until + * the event is delivered, and fails with the startup failure when startup + * fails. Writes after the machine stopped or completed fail with + * `StoppedError`. * * @since 0.4.0 */ readonly send: Atom.Writable< - AsyncResult.AsyncResult, + AsyncResult.AsyncResult, Event > /** - * Writable atom that stops the machine. Writes before startup completes fail - * with `NotReadyError`. + * Writable atom that stops the machine. + * + * **Details** + * + * A stop written while the machine is still starting is held behind earlier + * writes and stops the machine as soon as startup succeeds. * * @since 0.4.0 */ - readonly stop: Atom.Writable, void> + readonly stop: Atom.Writable, void> /** * Creates a reactive bridge for a directly owned child machine. @@ -144,6 +147,39 @@ export interface MachineAtom ChildMachineAtom } +/** + * Namespace containing type-level members associated with `MachineAtom`. + * + * @category models + * @since 0.42.0 + */ +export declare namespace MachineAtom { + /** + * Any machine bridge, including bridges for machines without events. + * + * @category utility types + * @since 0.42.0 + */ + export type Any = MachineAtom + + /** + * Extracts the state type of a machine bridge. + * + * @category utility types + * @since 0.42.0 + */ + export type State = A extends MachineAtom ? State + : never + + /** + * Extracts the event type accepted by a machine bridge. + * + * @category utility types + * @since 0.42.0 + */ + export type Event = A extends MachineAtom ? Event : never +} + type RefState = Ref extends Machine.MachineRef ? State : never type RefError = Ref extends Machine.MachineRef ? Error : never @@ -251,23 +287,31 @@ export interface ChildMachineAtom, + AsyncResult.AsyncResult, Machine.ChildMachine.Event > /** * Writable atom that stops the active child. * - * Writes fail with `ChildNotActiveError` while the child is inactive. + * **Details** + * + * A stop written while the parent machine is still starting is held behind + * earlier writes. Writes fail with `ChildNotActiveError` when the state that + * owns the child is not active. * * @since 0.4.0 */ readonly stop: Atom.Writable< - AsyncResult.AsyncResult, + AsyncResult.AsyncResult, void > /** @@ -282,6 +326,32 @@ export interface ChildMachineAtom ChildMachineAtom } +/** + * Namespace containing type-level members associated with `ChildMachineAtom`. + * + * @category models + * @since 0.42.0 + */ +export declare namespace ChildMachineAtom { + /** + * Any child machine bridge. + * + * @category utility types + * @since 0.42.0 + */ + export type Any = ChildMachineAtom + + /** + * Extracts the state type of the child selected by a child bridge. + * + * @category utility types + * @since 0.42.0 + */ + export type State = A extends ChildMachineAtom ? + RefState> + : never +} + type BridgeStartError = Bridge extends MachineAtom ? StartError : Bridge extends ChildMachineAtom ? StartError : never @@ -294,7 +364,7 @@ type BridgeStartError = Bridge extends MachineAtom | ChildMachineAtom, + Parent extends MachineAtom.Any | ChildMachineAtom.Any, Child extends Machine.ChildMachine.Any > = ChildMachineAtom> diff --git a/packages/effect-machine/test/unstable/reactivity/AtomMachineCommands.test.ts b/packages/effect-machine/test/unstable/reactivity/AtomMachineCommands.test.ts new file mode 100644 index 00000000..f1ef2dbc --- /dev/null +++ b/packages/effect-machine/test/unstable/reactivity/AtomMachineCommands.test.ts @@ -0,0 +1,247 @@ +import { assert, describe, it } from "@effect/vitest" +import { Cause, Data, Effect, Layer, Option, Schema, Stream } from "effect" +import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity" +import { Machine } from "../../../src/index.js" +import { AtomMachine } from "../../../src/unstable/reactivity/index.js" + +class Count extends Schema.TaggedClass("Count")("Count", { + value: Schema.Number +}) {} +class Idle extends Schema.TaggedClass("Idle")("Idle", {}) {} +class Push extends Schema.TaggedClass("Push")("Push", { + digit: Schema.Number +}) {} +class Activate extends Schema.TaggedClass("Activate")("Activate", {}) {} +class StartupFailed extends Data.TaggedError("StartupFailed")<{}> {} + +// Appends digits, so the final value records the delivery order. +const makeDigits = () => + Machine.make({ + root: Machine.state({ states: { Count } }), + events: Machine.eventsFromSchemas(Push) + }).handle({ + initial: { target: "Count", decoded: true, data: new Count({ value: 0 }) }, + states: { + Count: { + on: { + Push: { + target: "Count", + decoded: true, + data: ({ state, event }) => new Count({ value: state.value * 10 + event.digit }) + } + } + } + } + }) + +const Digits = Machine.child("digits", makeDigits()) + +const makeParent = (initial: "Count" | "Idle") => + Machine.make({ + children: { digits: Digits }, + root: Machine.state({ states: { Idle, Count } }), + events: Machine.eventsFromSchemas(Activate) + }).handle({ + initial: initial === "Count" + ? { target: "Count", decoded: true, data: new Count({ value: 0 }) } + : { target: "Idle", decoded: true, data: new Idle() }, + states: { + Idle: { + on: { + Activate: { target: "Count", decoded: true, data: () => new Count({ value: 0 }) } + } + }, + Count: { + invoke: { src: "digits" } + } + } + }) + +const slowRuntime = () => Atom.runtime(Layer.effectDiscard(Effect.sleep("20 millis"))) + +const makeRegistry = Effect.acquireRelease( + Effect.sync(() => AtomRegistry.make()), + (registry) => Effect.sync(() => registry.dispose()) +) + +const mount = (registry: AtomRegistry.AtomRegistry, atom: Atom.Atom) => + Effect.acquireRelease(Effect.sync(() => registry.mount(atom)), (release) => Effect.sync(release)) + +const waitFor = ( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom, + predicate: (value: A) => boolean +): Effect.Effect => + AtomRegistry.toStream(registry, atom).pipe( + Stream.filter(predicate), + Stream.runHead, + Effect.map(Option.getOrThrow) + ) + +const settled = ( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom> +): Effect.Effect> => + waitFor(registry, atom, (result) => !AsyncResult.isInitial(result) && !result.waiting) + +const failureOf = (result: AsyncResult.AsyncResult): E => { + assert(AsyncResult.isFailure(result)) + const error = Cause.findErrorOption(result.cause) + assert(Option.isSome(error)) + return error.value +} + +describe("AtomMachine commands", () => { + it.live("delivers an event written while the bound runtime is still building", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.bind(slowRuntime()).make(makeDigits()) + yield* mount(registry, bridge.result) + registry.set(bridge.send, new Push({ digit: 7 })) + const pending = registry.get(bridge.send) + assert(AsyncResult.isInitial(pending) && pending.waiting) + const state = yield* waitFor( + registry, + bridge.result, + (result) => AsyncResult.isSuccess(result) && result.value.state.value.value === 7 + ) + assert(AsyncResult.isSuccess(state)) + assert(AsyncResult.isSuccess(yield* settled(registry, bridge.send))) + }))) + + it.live("delivers several early events in write order", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.bind(slowRuntime()).make(makeDigits()) + yield* mount(registry, bridge.result) + registry.set(bridge.send, new Push({ digit: 1 })) + registry.set(bridge.send, new Push({ digit: 2 })) + registry.set(bridge.send, new Push({ digit: 3 })) + yield* waitFor( + registry, + bridge.result, + (result) => AsyncResult.isSuccess(result) && result.value.state.value.value === 123 + ) + // Writes made right after readiness still follow the held writes. + registry.set(bridge.send, new Push({ digit: 4 })) + yield* waitFor( + registry, + bridge.result, + (result) => AsyncResult.isSuccess(result) && result.value.state.value.value === 1234 + ) + }))) + + it.live("keeps writes after readiness ordered behind held writes", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.bind(slowRuntime()).make(makeDigits()) + yield* mount(registry, bridge.result) + registry.set(bridge.send, new Push({ digit: 1 })) + registry.set(bridge.send, new Push({ digit: 2 })) + yield* waitFor(registry, bridge.ref, AsyncResult.isSuccess) + registry.set(bridge.send, new Push({ digit: 3 })) + yield* waitFor( + registry, + bridge.result, + (result) => AsyncResult.isSuccess(result) && result.value.state.value.value === 123 + ) + }))) + + it.live("fails a write after the machine stopped with StoppedError", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.make(makeDigits()) + yield* mount(registry, bridge.snapshot) + yield* mount(registry, bridge.send) + registry.set(bridge.stop, undefined) + yield* waitFor( + registry, + bridge.snapshot, + (snapshot) => AsyncResult.isSuccess(snapshot) && snapshot.value.status === "stopped" + ) + registry.set(bridge.send, new Push({ digit: 1 })) + const result = yield* settled(registry, bridge.send) + assert.instanceOf(failureOf(result), Machine.StoppedError) + }))) + + it.live("settles early writes with the startup failure", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const runtime = Atom.runtime( + Layer.effectDiscard(Effect.sleep("20 millis").pipe(Effect.andThen(Effect.fail(new StartupFailed())))) + ) + const bridge = AtomMachine.bind(runtime).make(makeDigits()) + yield* mount(registry, bridge.result) + yield* mount(registry, bridge.send) + registry.set(bridge.send, new Push({ digit: 1 })) + const result = yield* settled(registry, bridge.send) + assert.instanceOf(failureOf(result), StartupFailed) + }))) + + it.live("stops the machine after startup when stop is written early", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const bridge = AtomMachine.bind(slowRuntime()).make(makeDigits()) + yield* mount(registry, bridge.snapshot) + yield* mount(registry, bridge.stop) + registry.set(bridge.stop, undefined) + const pending = registry.get(bridge.stop) + assert(AsyncResult.isInitial(pending) && pending.waiting) + yield* waitFor( + registry, + bridge.snapshot, + (snapshot) => AsyncResult.isSuccess(snapshot) && snapshot.value.status === "stopped" + ) + assert(AsyncResult.isSuccess(yield* settled(registry, bridge.stop))) + }))) + + it.live("delivers a child event written before the parent is ready", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const parent = AtomMachine.bind(slowRuntime()).make(makeParent("Count")) + const child = parent.child(Digits) + yield* mount(registry, child.result) + yield* mount(registry, child.send) + registry.set(child.send, new Push({ digit: 4 })) + registry.set(child.send, new Push({ digit: 2 })) + const pending = registry.get(child.send) + assert(AsyncResult.isInitial(pending) && pending.waiting) + yield* waitFor( + registry, + child.result, + (result) => + AsyncResult.isSuccess(result) && Option.isSome(result.value) && result.value.value.state.value.value === 42 + ) + assert(AsyncResult.isSuccess(yield* settled(registry, child.send))) + }))) + + it.live("delivers a child event written while the parent starts synchronously", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const parent = AtomMachine.make(makeParent("Count")) + const child = parent.child(Digits) + registry.set(child.send, new Push({ digit: 9 })) + yield* mount(registry, child.result) + yield* waitFor( + registry, + child.result, + (result) => + AsyncResult.isSuccess(result) && Option.isSome(result.value) && result.value.value.state.value.value === 9 + ) + }))) + + it.live("fails a child write with ChildNotActiveError when its owning state is inactive", () => + Effect.scoped(Effect.gen(function*() { + const registry = yield* makeRegistry + const parent = AtomMachine.bind(slowRuntime()).make(makeParent("Idle")) + const child = parent.child(Digits) + yield* mount(registry, parent.result) + yield* mount(registry, child.send) + yield* mount(registry, child.stop) + registry.set(child.send, new Push({ digit: 1 })) + const early = yield* settled(registry, child.send) + assert.instanceOf(failureOf(early), AtomMachine.ChildNotActiveError) + registry.set(child.stop, undefined) + assert.instanceOf(failureOf(registry.get(child.stop)), AtomMachine.ChildNotActiveError) + }))) +}) diff --git a/packages/effect-machine/typetest/unstable/reactivity/AtomMachine.tst.ts b/packages/effect-machine/typetest/unstable/reactivity/AtomMachine.tst.ts index 2000cc49..94842b17 100644 --- a/packages/effect-machine/typetest/unstable/reactivity/AtomMachine.tst.ts +++ b/packages/effect-machine/typetest/unstable/reactivity/AtomMachine.tst.ts @@ -644,4 +644,42 @@ describe("AtomMachine", () => { expect(bound.make).type.toBeCallableWith(complete) expect(bound.factory).type.toBeCallableWith(complete) }) + it("exposes generic helper types for machine and child bridges", () => { + const machine = makeMachine() + const bridge = AtomMachine.make(machine) + const eventless = AtomMachine.make( + Machine.make({ root: States, events: Machine.events({}) }).handle({ + initial: { target: "Idle", decoded: true, data: new Idle({}) }, + states: { Idle: {} } + }) + ) + expect(eventless).type.toBeAssignableTo() + expect(bridge).type.toBeAssignableTo() + expect>().type.toBe< + Machine.Machine.Snapshot> + >() + expect>().type.toBe>() + expect>().type.toBe() + expect>().type.toBe>() + + const Child = Machine.child("child", machine) + const parent = AtomMachine.make( + Machine.make({ children: { source1: Child }, root: States, events: Machine.eventsFromSchemas() }).handle({ + initial: { target: "Idle", decoded: true, data: new Idle({}) }, + states: { Idle: { invoke: { src: "source1" } } } + }) + ) + const child = parent.child(Child) + expect(child).type.toBeAssignableTo() + expect>().type.toBe< + Machine.Machine.Snapshot> + >() + }) + it("types early command results without a readiness failure", () => { + const bridge = AtomMachine.make(makeMachine()) + type StartError = Atom.Failure + expect>().type.toBe() + expect>().type.toBe() + expect(AtomMachine).type.not.toHaveProperty("NotReadyError") + }) })