From 7869157ad4b83d19c1b210043cadd11e7bc8e65f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 09:42:22 +0000 Subject: [PATCH 1/5] Bind started machines to the caller's Scope `Machine.start`, `Machine.resume`, and `prepared.start` now require `Scope.Scope`. Closing that scope stops the machine like `ref.stop`, and a machine that stops or completes earlier releases its place in the scope. `AtomMachine` is unaffected. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .changeset/scope-bound-machine-start.md | 19 +++ packages/effect-machine/README.md | 7 +- packages/effect-machine/docs/agent-guide.md | 7 +- .../effect-machine/perf/runtime/counter.mjs | 38 +++-- packages/effect-machine/src/Machine.ts | 40 +++-- .../src/internal/machine/runtime.ts | 103 +++++++++---- .../effect-machine/src/testing/MachineTest.ts | 4 +- .../internal/machine/ScopedLifetime.test.ts | 35 +++++ .../internal/machine/ScopedOwnership.test.ts | 45 ++++++ .../test/machine/ScopedLifetime.test.ts | 141 ++++++++++++++++++ .../machine/support/runtimeDifferential.ts | 4 +- .../typetest/machine/Resume.tst.ts | 4 +- .../typetest/machine/ScopedLifetime.tst.ts | 36 +++++ 13 files changed, 420 insertions(+), 63 deletions(-) create mode 100644 .changeset/scope-bound-machine-start.md create mode 100644 packages/effect-machine/test/internal/machine/ScopedLifetime.test.ts create mode 100644 packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts create mode 100644 packages/effect-machine/test/machine/ScopedLifetime.test.ts create mode 100644 packages/effect-machine/typetest/machine/ScopedLifetime.tst.ts diff --git a/.changeset/scope-bound-machine-start.md b/.changeset/scope-bound-machine-start.md new file mode 100644 index 00000000..c8082712 --- /dev/null +++ b/.changeset/scope-bound-machine-start.md @@ -0,0 +1,19 @@ +--- +"@typeonce/effect-machine": minor +--- + +Bind started machines to the caller's `Scope`. + +`Machine.start`, `Machine.resume`, and `prepared.start` now require `Scope.Scope`, like `Effect.forkScoped`. Closing that scope is equivalent to `ref.stop`: active invocations are interrupted, owned children stop, the snapshot becomes `stopped`, and later sends fail with `StoppedError`. A machine that stops or completes earlier releases its place in the scope, so long-lived scopes do not accumulate finalizers. For `prepared.start`, the scope that first evaluates it owns the machine. `AtomMachine` is unaffected. + +Wrap programs that start machines in `Effect.scoped`, or own a scope explicitly when a machine must outlive the caller: + +```ts +const program = Effect.scoped(Effect.gen(function*() { + const ref = yield* Machine.start(machine) + yield* ref.send(Events.Go()) +})) + +const scope = yield* Scope.make() +const detached = yield* Machine.start(machine).pipe(Scope.provide(scope)) +``` diff --git a/packages/effect-machine/README.md b/packages/effect-machine/README.md index 751f7a24..54c114d3 100644 --- a/packages/effect-machine/README.md +++ b/packages/effect-machine/README.md @@ -75,6 +75,11 @@ Root data is shared state. `update: "root"` replaces that value while retaining the active child configuration and its running work. Ordinary transitions target a descendant; the root itself is not a destination. +A started machine is owned by the enclosing `Scope`. Closing it is equivalent +to `ref.stop`: invocations are interrupted, owned children stop, and later +sends fail with `StoppedError`. Own a `Scope` explicitly when a machine must +outlive the program that started it. + ## Modeling workflow Put data on the lowest state that owns it. Use tagged schemas or `fields` for @@ -560,7 +565,7 @@ Snapshots can be validated for storage or transport: ```ts const encoded = yield * Machine.encodeSnapshot(machine, snapshot) const decoded = yield * Machine.decodeSnapshot(machine, encoded) -const ref = yield * Machine.resume(machine, decoded) +const ref = yield * Machine.resume(machine, decoded) // owned by the current Scope ``` Decoded snapshots are local runtime values and may contain class instances or diff --git a/packages/effect-machine/docs/agent-guide.md b/packages/effect-machine/docs/agent-guide.md index 44cc533e..3cfdc60b 100644 --- a/packages/effect-machine/docs/agent-guide.md +++ b/packages/effect-machine/docs/agent-guide.md @@ -81,14 +81,17 @@ through the exported descriptor: ```ts import { Effect } from "effect" -const program = Effect.gen(function*() { +const program = Effect.scoped(Effect.gen(function*() { const counter = yield* Machine.start(CounterMachine) yield* counter.send(CounterEvents.Start()) yield* counter.send(CounterEvents.Increment()) -}) +})) ``` +`Machine.start` requires a `Scope`. Closing it stops the machine and everything +it owns; use `ref.stop` to stop earlier. + ## Make impossible states unrepresentable A finite state describes how the machine behaves now. State data holds values diff --git a/packages/effect-machine/perf/runtime/counter.mjs b/packages/effect-machine/perf/runtime/counter.mjs index f51523ed..f6309399 100644 --- a/packages/effect-machine/perf/runtime/counter.mjs +++ b/packages/effect-machine/perf/runtime/counter.mjs @@ -24,7 +24,12 @@ if (machineRuntimePath === undefined) { throw new Error("Effect Machine benchmark could not locate the internal runtime module") } const machineRuntime = await import(pathToFileURL(machineRuntimePath).href) -const { Effect, Fiber, Option, Schema, Stream } = effect +const { Effect, Fiber, Option, Schema, Scope, Stream } = effect + +// Benchmarks own started machines explicitly and stop them through `ref.stop`; +// one long-lived Scope stands in for the enclosing program. +const benchmarkScope = Scope.makeUnsafe() +const startMachine = (machine, ...args) => Machine.start(machine, ...args).pipe(Scope.provide(benchmarkScope)) const invocationLifecycleMachine = benchmarkApi.effectLifecycle?.(Effect.void, Schema.Void) const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? [] : [true, false].map((enabled) => ({ @@ -35,7 +40,7 @@ const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? expected: () => 1, start: () => undefined, run: () => Effect.runPromise(Effect.gen(function*() { - const ref = yield* Machine.start(invocationLifecycleMachine) + const ref = yield* startMachine(invocationLifecycleMachine) yield* ref.join return 1 }).pipe(Effect.withTracerEnabled(enabled))), @@ -327,14 +332,14 @@ const planParallelCounterBatch = (size) => }) ) -export const startCounter = () => Effect.runPromise(Machine.start(counterMachine)) +export const startCounter = () => Effect.runPromise(startMachine(counterMachine)) export const stopCounter = (ref) => Effect.runPromise(ref.stop) export const startObservedCounter = () => Effect.runPromise( Effect.gen(function*() { - const ref = yield* Machine.start(counterMachine) + const ref = yield* startMachine(counterMachine) const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach) yield* Effect.yieldNow return { ref, observer } @@ -374,7 +379,7 @@ const waitForCounterChild = (parent) => export const startChildCounter = () => Effect.runPromise( Effect.gen(function*() { - const parent = yield* Machine.start(counterParentMachine) + const parent = yield* startMachine(counterParentMachine) yield* waitForCounterChild(parent) return parent }) @@ -412,13 +417,13 @@ export const runCounterBurst = (ref, size) => }) ) -const startHierarchicalCounter = () => Effect.runPromise(Machine.start(hierarchicalCounterMachine)) -const startParallelCounter = () => Effect.runPromise(Machine.start(parallelCounterMachine)) +const startHierarchicalCounter = () => Effect.runPromise(startMachine(hierarchicalCounterMachine)) +const startParallelCounter = () => Effect.runPromise(startMachine(parallelCounterMachine)) const startObservedHierarchicalCounter = () => Effect.runPromise( Effect.gen(function*() { - const ref = yield* Machine.start(hierarchicalCounterMachine) + const ref = yield* startMachine(hierarchicalCounterMachine) const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach) yield* Effect.yieldNow return { ref, observer } @@ -464,7 +469,7 @@ export const startCounters = (count) => Effect.runPromise( Effect.forEach( Array.from({ length: count }), - () => Machine.start(counterMachine), + () => startMachine(counterMachine), { concurrency: 1 } ) ) @@ -483,7 +488,7 @@ export const startChildCounters = (count) => Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* Machine.start(counterParentMachine) + const parent = yield* startMachine(counterParentMachine) yield* waitForCounterChild(parent) return parent }), @@ -525,7 +530,7 @@ const startRawProcesses = (count) => Effect.runPromise( Effect.forEach( Array.from({ length: count }), - () => machineRuntime.startProcess(rawProcessLogic), + () => machineRuntime.startProcess(rawProcessLogic).pipe(Scope.provide(benchmarkScope)), { concurrency: 1 } ) ) @@ -534,7 +539,10 @@ const startCompiledRawProcesses = (count) => Effect.runPromise( Effect.forEach( Array.from({ length: count }), - () => machineRuntime.startProcessWithStrategyForTesting(rawCompiledProcessLogic, "compiled"), + () => + machineRuntime.startProcessWithStrategyForTesting(rawCompiledProcessLogic, "compiled").pipe( + Scope.provide(benchmarkScope) + ), { concurrency: 1 } ) ) @@ -543,7 +551,7 @@ const startIndependentCounterPairs = (count) => Effect.runPromise( Effect.forEach( Array.from({ length: count }), - () => Effect.all([Machine.start(counterMachine), Machine.start(counterMachine)], { concurrency: 1 }), + () => Effect.all([startMachine(counterMachine), startMachine(counterMachine)], { concurrency: 1 }), { concurrency: 1 } ) ) @@ -556,7 +564,7 @@ const startObservedChildCounters = (count) => Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* Machine.start(counterParentMachine) + const parent = yield* startMachine(counterParentMachine) yield* waitForCounterChild(parent) const observer = yield* parent.childChanges(CounterChild).pipe( Stream.runDrain, @@ -575,7 +583,7 @@ const startSnapshotObservedChildCounters = (count) => Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* Machine.start(counterSnapshotParentMachine) + const parent = yield* startMachine(counterSnapshotParentMachine) yield* waitForCounterChild(parent) return parent }), diff --git a/packages/effect-machine/src/Machine.ts b/packages/effect-machine/src/Machine.ts index 97fdbc3c..16b8019b 100644 --- a/packages/effect-machine/src/Machine.ts +++ b/packages/effect-machine/src/Machine.ts @@ -1997,7 +1997,8 @@ export type RuntimeOutcome = * * Subscribe to `emissions` before evaluating `start` when initial-entry * emissions must be observed. `start` is one-shot: concurrent and repeated - * evaluations share the same initialization and running reference. + * evaluations share the same initialization and running reference, and the + * `Scope` of the first evaluation owns the machine. * * @category models * @since 0.11.0 @@ -2023,7 +2024,7 @@ export interface Prepared - /** Initializes this machine once and returns its running reference. */ + /** Initializes this machine once, owned by the first evaluating `Scope`, and returns its running reference. */ readonly start: Effect.Effect, StartError, StartRequirements> } @@ -9203,6 +9204,9 @@ export const waitFor: { * `prepared.start`. Ordinary callers can continue to use {@link start}, which * starts directly and does not allocate the prepared lifecycle boundary. * + * `prepared.start` is one-shot and scoped: the `Scope` that first evaluates it + * owns the machine, and closing that `Scope` is equivalent to `ref.stop`. + * * ```ts * const prepared = yield* Machine.prepare(machine) * @@ -9271,7 +9275,8 @@ export const prepare: < | MachineSchemaDecodeError | StartupError | StoppedError, - ExcludeCompatibleRuntime< + | Scope.Scope + | ExcludeCompatibleRuntime< ExecutionServices, Machine.EventOf, Machine.EmittedEventOf @@ -9280,7 +9285,7 @@ export const prepare: < > = internal.prepare /** - * Starts a machine. + * Starts a machine owned by the current `Scope`. * * **When to use** * @@ -9294,6 +9299,13 @@ export const prepare: < * publishes the new state, delivers emitted events, and then starts invokes * for entered states. * + * The machine is a scoped resource, like `Effect.forkScoped`. Closing the + * `Scope` is equivalent to `ref.stop`: active invocations are interrupted, + * owned children stop, the snapshot becomes `stopped`, and later sends fail + * with `StoppedError`. An earlier `ref.stop` or terminal completion leaves the + * finalizer with nothing to do. Own a `Scope` explicitly, for example with + * `Scope.make` and `Scope.provide`, when the machine must outlive the caller. + * * **Gotchas** * * The returned handle's `send` operation only enqueues events. Transition @@ -9320,10 +9332,10 @@ export const prepare: < * Idle: {} * } * }) - * const state = Effect.gen(function*() { + * const state = Effect.scoped(Effect.gen(function*() { * const ref = yield* Machine.start(machine) * return yield* ref.state - * }) + * })) * ``` * * @see {@link plan} for inspecting the same transition plan without executing it. @@ -9386,7 +9398,8 @@ export const start: < | MachineSchemaDecodeError | StartupError | StoppedError, - ExcludeCompatibleRuntime< + | Scope.Scope + | ExcludeCompatibleRuntime< ExecutionServices, Machine.EventOf, Machine.EmittedEventOf @@ -9394,10 +9407,14 @@ export const start: < > = internal.start /** - * Starts a fresh managed runtime from a decoded logical snapshot. + * Starts a fresh managed runtime from a decoded logical snapshot, owned by the + * current `Scope`. * * **Details** * + * The resumed machine has the same scoped lifetime as {@link start}: closing + * the `Scope` is equivalent to `ref.stop`. + * * `resume` validates and normalizes the supplied snapshot before publishing it * as the first state. It does not call the machine's initial function, replay * entry or transition actions, re-deliver raised or emitted events, or @@ -9436,12 +9453,12 @@ export const start: < * Idle: {} * } * }) - * const resumed = Effect.gen(function*() { + * const resumed = Effect.scoped(Effect.gen(function*() { * const initial = yield* Machine.planInitial(machine) * const encoded = yield* Machine.encodeSnapshot(machine, initial.state) * const snapshot = yield* Machine.decodeSnapshot(machine, encoded) * return yield* Machine.resume(machine, snapshot) - * }) + * })) * ``` * * @see {@link decodeSnapshot} for the schema and transport boundary. @@ -9498,7 +9515,8 @@ export const resume: < Machine.EmittedEventOf >, MachineSchemaDecodeError, - ExcludeCompatibleRuntime< + | Scope.Scope + | ExcludeCompatibleRuntime< ExecutionServices, Machine.EventOf, Machine.EmittedEventOf diff --git a/packages/effect-machine/src/internal/machine/runtime.ts b/packages/effect-machine/src/internal/machine/runtime.ts index ea1f71f7..48dd7f62 100644 --- a/packages/effect-machine/src/internal/machine/runtime.ts +++ b/packages/effect-machine/src/internal/machine/runtime.ts @@ -2,6 +2,8 @@ import * as Deferred from "effect/Deferred" import * as Effect from "effect/Effect" +import * as Exit from "effect/Exit" +import * as Scope from "effect/Scope" import * as Stream from "effect/Stream" import * as InspectionRuntime from "./inspectionRuntime.js" import { startCompactCompiledInternal, startCompiledSync } from "./runtimeCompiled.js" @@ -28,6 +30,37 @@ const makeRuntime = makeProcessRuntime(startLogicInternal, startCompiledSync) export type ProcessRuntimeStrategy = "generic" | "compiled" | "auto" +// A root machine is owned by the Scope that starts it, like `Effect.forkScoped`. +// Each machine gets an owner scope forked from the caller's Scope: closing the +// caller's Scope stops the machine, and termination closes the owner scope so +// it leaves its parent again. Owned children keep their process-local +// ownership. +const ownedByScope = ( + start: ( + ownership: Pick + ) => Effect.Effect, StartError, Requirements> +): Effect.Effect, StartError, Requirements | Scope.Scope> => + Effect.flatMap(Effect.scope, (parent) => + Effect.suspend(() => { + const owner = Scope.forkUnsafe(parent) + let terminated = false + const onStop = Effect.suspend(() => { + terminated = true + return Scope.close(owner, Exit.void) + }) + return Effect.uninterruptibleMask((restore) => + restore(start({ onStop })).pipe( + Effect.tap((ref) => + Scope.addFinalizer( + owner, + Effect.suspend(() => terminated ? Effect.void : ref.stop) + ) + ), + Effect.onExit((exit) => Exit.isFailure(exit) ? Scope.close(owner, exit) : Effect.void) + ) + ) + })) + const startProcessWithStrategy = Effect.fnUntraced(function*( logic: ProcessLogic, strategy: ProcessRuntimeStrategy, @@ -47,15 +80,17 @@ const startProcessWithStrategy = Effect.fnUntraced(function*( runtime } if (strategy === "generic") { - return yield* startGenericInternal(logic, internalOptions) + return yield* ownedByScope((ownership) => startGenericInternal(logic, { ...internalOptions, ...ownership })) } if (strategy === "compiled") { if (logic.execution?._tag !== "Compiled") { return yield* Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) } - return yield* startCompactCompiledInternal(logic, internalOptions) + return yield* ownedByScope((ownership) => + startCompactCompiledInternal(logic, { ...internalOptions, ...ownership }) + ) } - return yield* startLogicInternal(logic, internalOptions) + return yield* ownedByScope((ownership) => startLogicInternal(logic, { ...internalOptions, ...ownership })) }) /** @internal Test-only startup strategy selection. */ @@ -75,7 +110,7 @@ export const startProcessWithStrategyForTesting = < ): Effect.Effect< MachineRef, InitialError, - Requirements + Requirements | Scope.Scope > => startProcessWithStrategy(logic, strategy, options) as any export const startProcess: < @@ -93,7 +128,7 @@ export const startProcess: < ) => Effect.Effect< MachineRef, InitialError, - Requirements + Requirements | Scope.Scope > = Effect.fnUntraced(function*( logic: ProcessLogic, options?: { @@ -101,18 +136,22 @@ export const startProcess: < } ) { const runtime = yield* makeRuntime - return yield* startLogicInternal( - logic, - options === undefined - ? { - detached: true, - runtime - } - : { - ...options, - detached: true, - runtime - } + return yield* ownedByScope((ownership) => + startLogicInternal( + logic, + options === undefined + ? { + detached: true, + runtime, + ...ownership + } + : { + ...options, + detached: true, + runtime, + ...ownership + } + ) ) }) @@ -155,17 +194,25 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< inspectionRoot: true, origin: { _tag: "Root" } } - const initialize = strategy === "generic" - ? startGenericInternal(logic, internalOptions) - : strategy === "compiled" - ? logic.execution?._tag === "Compiled" - ? startCompactCompiledInternal(logic, internalOptions) - : Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) - : startLogicInternal(logic, internalOptions) + const initialize = (ownership: Pick) => { + const startOptions = { ...internalOptions, ...ownership } + return strategy === "generic" + ? startGenericInternal(logic, startOptions) + : strategy === "compiled" + ? logic.execution?._tag === "Compiled" + ? startCompactCompiledInternal(logic, startOptions) + : Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) + : startLogicInternal(logic, startOptions) + } + // `start` is one-shot: the Scope that first evaluates it owns the machine. const start = yield* Effect.cached( - initialize.pipe( + ownedByScope(initialize).pipe( Effect.onExit((exit) => Deferred.done(started, exit)) - ) as Effect.Effect, InitialError, Requirements> + ) as Effect.Effect< + MachineRef, + InitialError, + Requirements | Scope.Scope + > ) return { id: options?.id ?? sessionId, @@ -193,7 +240,7 @@ export const prepareProcess: < readonly id?: string } ) => Effect.Effect< - PreparedProcess + PreparedProcess > = ((logic: ProcessLogic, options?: { readonly id?: string }) => prepareProcessWithStrategy(logic, "auto", options)) as any @@ -214,5 +261,5 @@ export const prepareProcessWithStrategyForTesting = < readonly id?: string } ): Effect.Effect< - PreparedProcess + PreparedProcess > => prepareProcessWithStrategy(logic, strategy, options) as any diff --git a/packages/effect-machine/src/testing/MachineTest.ts b/packages/effect-machine/src/testing/MachineTest.ts index e609b896..f04a05eb 100644 --- a/packages/effect-machine/src/testing/MachineTest.ts +++ b/packages/effect-machine/src/testing/MachineTest.ts @@ -511,10 +511,10 @@ export { ProbeUnavailableError } from "../internal/testing/machine/probe.js" * Idle: {} * } * }) - * const program = Effect.gen(function*() { + * const program = Effect.scoped(Effect.gen(function*() { * const ref = yield* Machine.start(machine) * return yield* MachineTest.probe(machine, ref) - * }) + * })) * ``` * * @category runtime testing diff --git a/packages/effect-machine/test/internal/machine/ScopedLifetime.test.ts b/packages/effect-machine/test/internal/machine/ScopedLifetime.test.ts new file mode 100644 index 00000000..f359973e --- /dev/null +++ b/packages/effect-machine/test/internal/machine/ScopedLifetime.test.ts @@ -0,0 +1,35 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Ref, Stream } from "effect" +import { Machine } from "../../../src/index.js" +import * as Process from "../../../src/internal/machine/process.js" +import type * as Runtime from "../../../src/internal/machine/runtime.js" + +const makeWatcher = (released: Ref.Ref) => + Machine.make({ + root: Machine.state({ states: { Watching: {} } }), + events: Machine.events({}), + streams: { watch: Stream.never.pipe(Stream.ensuring(Ref.set(released, true))) } + }).handle({ + initial: { target: "Watching" }, + states: { Watching: { invoke: { src: "watch", onDone: { none: true } } } } + }) + +const strategies: ReadonlyArray = ["generic", "compiled"] + +describe("Machine scoped lifetime strategies", () => { + for (const strategy of strategies) { + it.effect(`${strategy} runtime releases state-owned work on scope closure`, () => + Effect.gen(function*() { + const released = yield* Ref.make(false) + const machine = makeWatcher(released) + const ref = yield* Effect.scoped(Process.startWithRuntimeStrategyForTesting(machine, strategy)) + assert.strictEqual(yield* Ref.get(released), true) + assert.strictEqual((yield* ref.snapshot).status, "stopped") + + const prepared = yield* Process.prepareWithRuntimeStrategyForTesting(makeWatcher(released), strategy) + yield* Ref.set(released, false) + yield* Effect.scoped(prepared.start) + assert.strictEqual(yield* Ref.get(released), true) + })) + } +}) diff --git a/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts b/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts new file mode 100644 index 00000000..9c4c5a13 --- /dev/null +++ b/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts @@ -0,0 +1,45 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Exit, Scope } from "effect" +import { Machine } from "../../../src/index.js" + +const Events = Machine.events({ Finish: {} }) +const machine = Machine.make({ + root: Machine.state({ states: { Idle: {}, Done: { type: "final" } } }), + events: Events +}).handle({ + initial: { target: "Idle" }, + states: { Idle: { on: { Finish: { target: "Done" } } }, Done: {} } +}) + +type ScopeState = { readonly _tag: string; readonly finalizers?: Map } + +const finalizerCount = (scope: Scope.Closeable): number => { + const state = (scope as unknown as { readonly state: ScopeState }).state + return state._tag === "Empty" || state._tag === "Closed" ? 0 : state.finalizers?.size ?? 1 +} + +describe("Machine scoped ownership", () => { + it.effect("terminated machines leave the owning scope", () => + Effect.gen(function*() { + const owner = yield* Scope.make() + const stopped = yield* Machine.start(machine).pipe(Scope.provide(owner)) + const completed = yield* Machine.start(machine).pipe(Scope.provide(owner)) + const running = yield* Machine.start(machine).pipe(Scope.provide(owner)) + assert.strictEqual(finalizerCount(owner), 3) + yield* stopped.stop + yield* completed.send(Events.Finish()) + yield* completed.join + assert.strictEqual(finalizerCount(owner), 1) + yield* Scope.close(owner, Exit.void) + assert.strictEqual((yield* running.snapshot).status, "stopped") + assert.strictEqual(finalizerCount(owner), 0) + })) + + it.effect("starting inside a closed scope stops the machine immediately", () => + Effect.gen(function*() { + const owner = yield* Scope.make() + yield* Scope.close(owner, Exit.void) + const ref = yield* Machine.start(machine).pipe(Scope.provide(owner)) + assert.strictEqual((yield* ref.snapshot).status, "stopped") + })) +}) diff --git a/packages/effect-machine/test/machine/ScopedLifetime.test.ts b/packages/effect-machine/test/machine/ScopedLifetime.test.ts new file mode 100644 index 00000000..de2ce34c --- /dev/null +++ b/packages/effect-machine/test/machine/ScopedLifetime.test.ts @@ -0,0 +1,141 @@ +import { assert, describe, it } from "@effect/vitest" +import { Deferred, Effect, Exit, Option, Ref, Scope, Stream } from "effect" +import { Machine } from "../../src/index.js" + +const Events = Machine.events({ Go: {} }) + +const makeWatcher = (released: Ref.Ref) => + Machine.make({ + root: Machine.state({ states: { Watching: {} } }), + events: Events, + streams: { watch: Stream.never.pipe(Stream.ensuring(Ref.set(released, true))) } + }).handle({ + initial: { target: "Watching" }, + states: { Watching: { invoke: { src: "watch", onDone: { none: true } } } } + }) + +describe("Machine scoped lifetime", () => { + it.effect("closing the starting scope stops the machine and releases state-owned work", () => + Effect.gen(function*() { + const released = yield* Ref.make(false) + const machine = makeWatcher(released) + const ref = yield* Effect.scoped(Machine.start(machine)) + assert.strictEqual(yield* Ref.get(released), true) + const snapshot = yield* ref.snapshot + assert.strictEqual(snapshot.status, "stopped") + const sent = yield* Effect.flip(ref.send(Events.Go())) + assert.instanceOf(sent, Machine.StoppedError) + })) + + it.effect("interrupts effect invocations and stops invoked children", () => + Effect.gen(function*() { + const effectInterrupted = yield* Deferred.make() + const child = Machine.make({ + root: Machine.state({ states: { Running: {} } }), + events: Machine.events({}) + }).handle({ + initial: { target: "Running" }, + states: { Running: {} } + }) + const Child = Machine.child("child", child) + const parent = Machine.make({ + root: Machine.state({ states: { Active: {} } }), + events: Machine.events({}), + effects: { hold: Effect.never.pipe(Effect.onInterrupt(() => Deferred.succeed(effectInterrupted, void 0))) }, + children: { child: Child } + }).handle({ + initial: { target: "Active" }, + states: { + Active: { + invoke: [ + { src: "hold" }, + { src: "child" } + ] + } + } + }) + const { childRef, ref } = yield* Effect.scoped(Effect.gen(function*() { + const ref = yield* Machine.start(parent) + const childRef = yield* ref.childChanges(Child).pipe( + Stream.filter(Option.isSome), + Stream.runHead, + Effect.map(Option.flatten), + Effect.map(Option.getOrThrow) + ) + assert.strictEqual((yield* childRef.snapshot).status, "active") + return { childRef, ref } + })) + yield* Deferred.await(effectInterrupted) + assert.strictEqual((yield* ref.snapshot).status, "stopped") + assert.strictEqual((yield* childRef.snapshot).status, "stopped") + })) + + it.effect("an earlier stop leaves the scope finalizer with nothing to do", () => + Effect.gen(function*() { + const released = yield* Ref.make(false) + const machine = makeWatcher(released) + const ref = yield* Effect.scoped(Effect.gen(function*() { + const ref = yield* Machine.start(machine) + yield* ref.stop + assert.strictEqual(yield* Ref.get(released), true) + return ref + })) + assert.strictEqual((yield* ref.snapshot).status, "stopped") + })) + + it.effect("a completed machine is unaffected by scope closure", () => + Effect.gen(function*() { + const machine = Machine.make({ + root: Machine.state({ states: { Done: { type: "final" } } }), + events: Machine.events({}) + }).handle({ initial: { target: "Done" }, states: { Done: {} } }) + const ref = yield* Effect.scoped(Effect.gen(function*() { + const ref = yield* Machine.start(machine) + yield* ref.join + return ref + })) + assert.strictEqual((yield* ref.snapshot).status, "done") + })) + + it.effect("resume and prepared.start share the scoped lifetime", () => + Effect.gen(function*() { + const resumedReleased = yield* Ref.make(false) + const resumedMachine = makeWatcher(resumedReleased) + const initial = yield* Machine.planInitial(resumedMachine) + const resumed = yield* Effect.scoped(Machine.resume(resumedMachine, initial.state)) + assert.strictEqual(yield* Ref.get(resumedReleased), true) + assert.strictEqual((yield* resumed.snapshot).status, "stopped") + + const preparedReleased = yield* Ref.make(false) + const preparedMachine = makeWatcher(preparedReleased) + const prepared = yield* Machine.prepare(preparedMachine) + const started = yield* Effect.scoped(prepared.start) + assert.strictEqual(yield* Ref.get(preparedReleased), true) + assert.strictEqual((yield* started.snapshot).status, "stopped") + })) + + it.effect("the scope that first evaluates prepared.start owns the machine", () => + Effect.gen(function*() { + const released = yield* Ref.make(false) + const prepared = yield* Machine.prepare(makeWatcher(released)) + const owner = yield* Scope.make() + const ref = yield* prepared.start.pipe(Scope.provide(owner)) + const again = yield* Effect.scoped(prepared.start) + assert.strictEqual(again, ref) + assert.strictEqual(yield* Ref.get(released), false) + yield* Scope.close(owner, Exit.void) + assert.strictEqual(yield* Ref.get(released), true) + })) + + it.effect("an explicitly owned scope lets a machine outlive its caller", () => + Effect.gen(function*() { + const released = yield* Ref.make(false) + const machine = makeWatcher(released) + const owner = yield* Scope.make() + const ref = yield* Effect.scoped(Machine.start(machine).pipe(Scope.provide(owner))) + assert.strictEqual(yield* Ref.get(released), false) + assert.strictEqual((yield* ref.snapshot).status, "active") + yield* Scope.close(owner, Exit.void) + assert.strictEqual(yield* Ref.get(released), true) + })) +}) diff --git a/packages/effect-machine/test/machine/support/runtimeDifferential.ts b/packages/effect-machine/test/machine/support/runtimeDifferential.ts index 72f994d7..6637302f 100644 --- a/packages/effect-machine/test/machine/support/runtimeDifferential.ts +++ b/packages/effect-machine/test/machine/support/runtimeDifferential.ts @@ -1,5 +1,5 @@ import { assert } from "@effect/vitest" -import { Cause, Effect, Fiber, Stream } from "effect" +import { Cause, Effect, Fiber, type Scope, Stream } from "effect" import { isDeepStrictEqual } from "node:util" import { Machine } from "../../../src/index.js" import { MachineTest } from "../../../src/testing/index.js" @@ -63,7 +63,7 @@ const waitForBoundary = ( */ export const verifyManagedExecution = Effect.fn(function*(options: { readonly machine: Machine.Machine.Any - readonly open: Effect.Effect, unknown, never> + readonly open: Effect.Effect, unknown, Scope.Scope> readonly initial: DifferentialBoundary readonly steps: ReadonlyArray readonly label: string diff --git a/packages/effect-machine/typetest/machine/Resume.tst.ts b/packages/effect-machine/typetest/machine/Resume.tst.ts index 6837c389..b7a5f1dd 100644 --- a/packages/effect-machine/typetest/machine/Resume.tst.ts +++ b/packages/effect-machine/typetest/machine/Resume.tst.ts @@ -1,4 +1,4 @@ -import { Effect, Schema } from "effect" +import { Effect, Schema, Scope } from "effect" import { Atom } from "effect/reactivity" import { describe, expect, it } from "tstyche" import { Machine } from "../../src/index.js" @@ -41,7 +41,7 @@ describe("Machine logical resumption", () => { const resumed = Machine.resume(machine, snapshot) type Ref = Effect.Success expect>().type.toBe() - expect>().type.toBe() + expect>().type.toBe() expect>().type.toBe< Machine.InfiniteTransitionError | Machine.MachineSchemaDecodeError | Machine.StoppedError >() diff --git a/packages/effect-machine/typetest/machine/ScopedLifetime.tst.ts b/packages/effect-machine/typetest/machine/ScopedLifetime.tst.ts new file mode 100644 index 00000000..436181a3 --- /dev/null +++ b/packages/effect-machine/typetest/machine/ScopedLifetime.tst.ts @@ -0,0 +1,36 @@ +import { Effect, Schema, Scope } from "effect" +import { describe, expect, it } from "tstyche" +import { Machine } from "../../src/index.js" +import { AtomMachine } from "../../src/unstable/reactivity/index.js" + +class Idle extends Schema.TaggedClass("Idle")("Idle", {}) {} +const machine = Machine.make({ + root: Machine.state({ states: { Idle } }), + events: Machine.eventsFromSchemas() +}).handle({ + initial: { target: "Idle", decoded: true, data: new Idle({}) }, + states: { Idle: {} } +}) + +describe("Machine scoped lifetime", () => { + it("requires a Scope to start, resume, and prepare", () => { + const started = Machine.start(machine) + expect>().type.toBe() + const resumed = Machine.resume(machine, { + path: "" as const, + value: undefined, + state: { path: "Idle", value: new Idle({}) } + }) + expect>().type.toBe() + const prepared = Machine.prepare(machine) + expect>().type.toBe() + expect["start"]>>().type.toBe() + expect(Effect.scoped(started)).type.toBe< + Effect.Effect, Effect.Error, never> + >() + }) + + it("keeps machines acceptable by the Atom bridge", () => { + expect(AtomMachine.make(machine)).type.not.toRaiseError() + }) +}) From a3ba68cdfb1902243e4094e97cd69be6e5c9d63c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 09:43:35 +0000 Subject: [PATCH 2/5] Format scoped start changes Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .changeset/scope-bound-machine-start.md | 6 ++++-- packages/effect-machine/src/internal/machine/runtime.ts | 4 +--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.changeset/scope-bound-machine-start.md b/.changeset/scope-bound-machine-start.md index c8082712..369777b5 100644 --- a/.changeset/scope-bound-machine-start.md +++ b/.changeset/scope-bound-machine-start.md @@ -14,6 +14,8 @@ const program = Effect.scoped(Effect.gen(function*() { yield* ref.send(Events.Go()) })) -const scope = yield* Scope.make() -const detached = yield* Machine.start(machine).pipe(Scope.provide(scope)) +const detached = Effect.gen(function*() { + const scope = yield* Scope.make() + return yield* Machine.start(machine).pipe(Scope.provide(scope)) +}) ``` diff --git a/packages/effect-machine/src/internal/machine/runtime.ts b/packages/effect-machine/src/internal/machine/runtime.ts index 48dd7f62..43dfaf53 100644 --- a/packages/effect-machine/src/internal/machine/runtime.ts +++ b/packages/effect-machine/src/internal/machine/runtime.ts @@ -86,9 +86,7 @@ const startProcessWithStrategy = Effect.fnUntraced(function*( if (logic.execution?._tag !== "Compiled") { return yield* Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) } - return yield* ownedByScope((ownership) => - startCompactCompiledInternal(logic, { ...internalOptions, ...ownership }) - ) + return yield* ownedByScope((ownership) => startCompactCompiledInternal(logic, { ...internalOptions, ...ownership })) } return yield* ownedByScope((ownership) => startLogicInternal(logic, { ...internalOptions, ...ownership })) }) From 6d7d55463d23e6b55ac8c41c2b54f6f8b736acc8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 09:54:15 +0000 Subject: [PATCH 3/5] Keep scope ownership of started machines lightweight Fork the owner scope only after a successful start, register the ref's own stop effect, and close the owner scope synchronously on termination. This avoids a per-machine copy of the services map and per-start closures. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .../src/internal/machine/runtime.ts | 37 +++++++++---------- 1 file changed, 17 insertions(+), 20 deletions(-) diff --git a/packages/effect-machine/src/internal/machine/runtime.ts b/packages/effect-machine/src/internal/machine/runtime.ts index 43dfaf53..87492966 100644 --- a/packages/effect-machine/src/internal/machine/runtime.ts +++ b/packages/effect-machine/src/internal/machine/runtime.ts @@ -31,33 +31,30 @@ const makeRuntime = makeProcessRuntime(startLogicInternal, startCompiledSync) export type ProcessRuntimeStrategy = "generic" | "compiled" | "auto" // A root machine is owned by the Scope that starts it, like `Effect.forkScoped`. -// Each machine gets an owner scope forked from the caller's Scope: closing the -// caller's Scope stops the machine, and termination closes the owner scope so -// it leaves its parent again. Owned children keep their process-local -// ownership. +// Each started machine gets an owner scope forked from the caller's Scope: +// closing the caller's Scope stops the machine, and termination closes the +// owner scope so it leaves its parent again. The owner scope holds only the +// ref's own `stop` effect, so termination can drop the closing effect. +// Owned children keep their process-local ownership. const ownedByScope = ( start: ( - ownership: Pick + ownership: Pick ) => Effect.Effect, StartError, Requirements> ): Effect.Effect, StartError, Requirements | Scope.Scope> => Effect.flatMap(Effect.scope, (parent) => Effect.suspend(() => { - const owner = Scope.forkUnsafe(parent) + let owner: Scope.Closeable | undefined let terminated = false - const onStop = Effect.suspend(() => { + const onStopSync = () => { terminated = true - return Scope.close(owner, Exit.void) - }) - return Effect.uninterruptibleMask((restore) => - restore(start({ onStop })).pipe( - Effect.tap((ref) => - Scope.addFinalizer( - owner, - Effect.suspend(() => terminated ? Effect.void : ref.stop) - ) - ), - Effect.onExit((exit) => Exit.isFailure(exit) ? Scope.close(owner, exit) : Effect.void) - ) + if (owner !== undefined) Scope.closeUnsafe(owner, Exit.void) + } + return Effect.uninterruptible( + Effect.flatMap(start({ onStopSync }), (ref) => { + if (terminated) return Effect.succeed(ref) + owner = Scope.forkUnsafe(parent) + return Effect.as(Scope.addFinalizer(owner, ref.stop), ref) + }) ) })) @@ -192,7 +189,7 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< inspectionRoot: true, origin: { _tag: "Root" } } - const initialize = (ownership: Pick) => { + const initialize = (ownership: Pick) => { const startOptions = { ...internalOptions, ...ownership } return strategy === "generic" ? startGenericInternal(logic, startOptions) From 75c285afa7570f707f08cc5100c8a9b3d8582068 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 10:04:46 +0000 Subject: [PATCH 4/5] Track scope-owned machines in one record per scope Each owning Scope keeps one finalizer and a live set. The runtime registers a machine when it is ready and releases it synchronously when it terminates, so ownership costs one set entry per machine. Benchmarks start machines through one shared service context with a long-lived scope. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .../effect-machine/perf/runtime/counter.mjs | 84 +++++++++-------- .../src/internal/machine/runtime.ts | 91 +++++++++++++------ .../src/internal/machine/runtimeCompiled.ts | 4 +- .../src/internal/machine/runtimeGeneric.ts | 2 +- .../src/internal/machine/runtimeProtocol.ts | 3 +- .../internal/machine/ScopedOwnership.test.ts | 26 +++++- 6 files changed, 130 insertions(+), 80 deletions(-) diff --git a/packages/effect-machine/perf/runtime/counter.mjs b/packages/effect-machine/perf/runtime/counter.mjs index f6309399..a6d27f5b 100644 --- a/packages/effect-machine/perf/runtime/counter.mjs +++ b/packages/effect-machine/perf/runtime/counter.mjs @@ -24,12 +24,13 @@ if (machineRuntimePath === undefined) { throw new Error("Effect Machine benchmark could not locate the internal runtime module") } const machineRuntime = await import(pathToFileURL(machineRuntimePath).href) -const { Effect, Fiber, Option, Schema, Scope, Stream } = effect +const { Context, Effect, Fiber, Option, Schema, Scope, Stream } = effect -// Benchmarks own started machines explicitly and stop them through `ref.stop`; -// one long-lived Scope stands in for the enclosing program. +// Benchmarks own started machines explicitly and stop them through `ref.stop`. +// One long-lived Scope in one shared service context stands in for the +// enclosing program, so starts do not copy the service context per machine. const benchmarkScope = Scope.makeUnsafe() -const startMachine = (machine, ...args) => Machine.start(machine, ...args).pipe(Scope.provide(benchmarkScope)) +const runPromise = Effect.runPromiseWith(Context.make(Scope.Scope, benchmarkScope)) const invocationLifecycleMachine = benchmarkApi.effectLifecycle?.(Effect.void, Schema.Void) const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? [] : [true, false].map((enabled) => ({ @@ -39,8 +40,8 @@ const invocationLifecycleBenchmarks = invocationLifecycleMachine === undefined ? operations: () => 1, expected: () => 1, start: () => undefined, - run: () => Effect.runPromise(Effect.gen(function*() { - const ref = yield* startMachine(invocationLifecycleMachine) + run: () => runPromise(Effect.gen(function*() { + const ref = yield* Machine.start(invocationLifecycleMachine) yield* ref.join return 1 }).pipe(Effect.withTracerEnabled(enabled))), @@ -332,14 +333,14 @@ const planParallelCounterBatch = (size) => }) ) -export const startCounter = () => Effect.runPromise(startMachine(counterMachine)) +export const startCounter = () => runPromise(Machine.start(counterMachine)) -export const stopCounter = (ref) => Effect.runPromise(ref.stop) +export const stopCounter = (ref) => runPromise(ref.stop) export const startObservedCounter = () => - Effect.runPromise( + runPromise( Effect.gen(function*() { - const ref = yield* startMachine(counterMachine) + const ref = yield* Machine.start(counterMachine) const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach) yield* Effect.yieldNow return { ref, observer } @@ -347,12 +348,12 @@ export const startObservedCounter = () => ) export const stopObservedCounter = ({ ref, observer }) => - Effect.runPromise( + runPromise( ref.stop.pipe(Effect.ensuring(Fiber.interrupt(observer))) ) export const runObservedCounterBurst = ({ ref, observer }, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { yield* ref.send(incrementEvent) @@ -377,18 +378,18 @@ const waitForCounterChild = (parent) => }) export const startChildCounter = () => - Effect.runPromise( + runPromise( Effect.gen(function*() { - const parent = yield* startMachine(counterParentMachine) + const parent = yield* Machine.start(counterParentMachine) yield* waitForCounterChild(parent) return parent }) ) -export const stopChildCounter = (parent) => Effect.runPromise(parent.stop) +export const stopChildCounter = (parent) => runPromise(parent.stop) export const runChildCounterBurst = (parent, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { const child = yield* parent.child(CounterChild) @@ -407,7 +408,7 @@ export const runChildCounterBurst = (parent, size) => ) export const runCounterBurst = (ref, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { yield* ref.send(incrementEvent) @@ -417,13 +418,13 @@ export const runCounterBurst = (ref, size) => }) ) -const startHierarchicalCounter = () => Effect.runPromise(startMachine(hierarchicalCounterMachine)) -const startParallelCounter = () => Effect.runPromise(startMachine(parallelCounterMachine)) +const startHierarchicalCounter = () => runPromise(Machine.start(hierarchicalCounterMachine)) +const startParallelCounter = () => runPromise(Machine.start(parallelCounterMachine)) const startObservedHierarchicalCounter = () => - Effect.runPromise( + runPromise( Effect.gen(function*() { - const ref = yield* startMachine(hierarchicalCounterMachine) + const ref = yield* Machine.start(hierarchicalCounterMachine) const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach) yield* Effect.yieldNow return { ref, observer } @@ -431,7 +432,7 @@ const startObservedHierarchicalCounter = () => ) const runHierarchicalCounterBurst = (ref, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { yield* ref.send(hierarchicalIncrementEvent) @@ -442,7 +443,7 @@ const runHierarchicalCounterBurst = (ref, size) => ) const runParallelCounterBurst = (ref, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { yield* ref.send(index % 2 === 0 ? parallelIncrementLeftEvent : parallelIncrementRightEvent) @@ -453,7 +454,7 @@ const runParallelCounterBurst = (ref, size) => ) const runObservedHierarchicalCounterBurst = ({ ref, observer }, size) => - Effect.runPromise( + runPromise( Effect.gen(function*() { for (let index = 0; index < size; index += 1) { yield* ref.send(hierarchicalIncrementEvent) @@ -466,16 +467,16 @@ const runObservedHierarchicalCounterBurst = ({ ref, observer }, size) => ) export const startCounters = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), - () => startMachine(counterMachine), + () => Machine.start(counterMachine), { concurrency: 1 } ) ) export const stopCounters = (refs) => - Effect.runPromise( + runPromise( Effect.forEach(refs, (ref) => ref.stop, { concurrency: "unbounded", discard: true @@ -483,12 +484,12 @@ export const stopCounters = (refs) => ) export const startChildCounters = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* startMachine(counterParentMachine) + const parent = yield* Machine.start(counterParentMachine) yield* waitForCounterChild(parent) return parent }), @@ -527,31 +528,28 @@ const rawCompiledProcessLogic = machineRuntime.compiledProcess === undefined } const startRawProcesses = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), - () => machineRuntime.startProcess(rawProcessLogic).pipe(Scope.provide(benchmarkScope)), + () => machineRuntime.startProcess(rawProcessLogic), { concurrency: 1 } ) ) const startCompiledRawProcesses = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), - () => - machineRuntime.startProcessWithStrategyForTesting(rawCompiledProcessLogic, "compiled").pipe( - Scope.provide(benchmarkScope) - ), + () => machineRuntime.startProcessWithStrategyForTesting(rawCompiledProcessLogic, "compiled"), { concurrency: 1 } ) ) const startIndependentCounterPairs = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), - () => Effect.all([startMachine(counterMachine), startMachine(counterMachine)], { concurrency: 1 }), + () => Effect.all([Machine.start(counterMachine), Machine.start(counterMachine)], { concurrency: 1 }), { concurrency: 1 } ) ) @@ -559,12 +557,12 @@ const startIndependentCounterPairs = (count) => const stopIndependentCounterPairs = (pairs) => stopCounters(pairs.flat()) const startObservedChildCounters = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* startMachine(counterParentMachine) + const parent = yield* Machine.start(counterParentMachine) yield* waitForCounterChild(parent) const observer = yield* parent.childChanges(CounterChild).pipe( Stream.runDrain, @@ -578,12 +576,12 @@ const startObservedChildCounters = (count) => ) const startSnapshotObservedChildCounters = (count) => - Effect.runPromise( + runPromise( Effect.forEach( Array.from({ length: count }), () => Effect.gen(function*() { - const parent = yield* startMachine(counterSnapshotParentMachine) + const parent = yield* Machine.start(counterSnapshotParentMachine) yield* waitForCounterChild(parent) return parent }), @@ -592,7 +590,7 @@ const startSnapshotObservedChildCounters = (count) => ) const stopObservedChildCounters = (units) => - Effect.runPromise( + runPromise( Effect.forEach( units, ({ parent, observer }) => parent.stop.pipe(Effect.ensuring(Fiber.interrupt(observer))), diff --git a/packages/effect-machine/src/internal/machine/runtime.ts b/packages/effect-machine/src/internal/machine/runtime.ts index 87492966..aba858e7 100644 --- a/packages/effect-machine/src/internal/machine/runtime.ts +++ b/packages/effect-machine/src/internal/machine/runtime.ts @@ -2,7 +2,6 @@ import * as Deferred from "effect/Deferred" import * as Effect from "effect/Effect" -import * as Exit from "effect/Exit" import * as Scope from "effect/Scope" import * as Stream from "effect/Stream" import * as InspectionRuntime from "./inspectionRuntime.js" @@ -31,32 +30,64 @@ const makeRuntime = makeProcessRuntime(startLogicInternal, startCompiledSync) export type ProcessRuntimeStrategy = "generic" | "compiled" | "auto" // A root machine is owned by the Scope that starts it, like `Effect.forkScoped`. -// Each started machine gets an owner scope forked from the caller's Scope: -// closing the caller's Scope stops the machine, and termination closes the -// owner scope so it leaves its parent again. The owner scope holds only the -// ref's own `stop` effect, so termination can drop the closing effect. -// Owned children keep their process-local ownership. +// Each owning Scope carries one ownership record with a single finalizer that +// stops the machines still alive, in reverse start order. The record's shared +// hooks register a machine when it is ready and release it synchronously when +// it terminates, so machines cost one set entry and long-lived scopes do not +// accumulate finalizers. Owned children keep their process-local ownership. +type OwnershipHooks = Pick + +interface ScopeOwnership { + readonly live: Set> + readonly hooks: OwnershipHooks + closed: boolean +} + +const ownerships = new WeakMap() + +const stopLive = (ownership: ScopeOwnership): Effect.Effect => + Effect.suspend(() => { + ownership.closed = true + const refs = Array.from(ownership.live).reverse() + ownership.live.clear() + return Effect.forEach(refs, (ref) => ref.stop, { discard: true }) + }) + +const ownershipOf = (parent: Scope.Scope): Effect.Effect => + Effect.suspend(() => { + const existing = ownerships.get(parent) + if (existing !== undefined) return Effect.succeed(existing) + const live = new Set>() + const ownership: ScopeOwnership = { + live, + hooks: { + onReadySync: (ref) => { + live.add(ref) + return true + }, + onStopSync: (ref) => { + live.delete(ref) + } + }, + closed: false + } + ownerships.set(parent, ownership) + // A closed Scope runs the finalizer immediately, which marks the record + // closed so machines started afterwards stop right away. + return Effect.as(Scope.addFinalizer(parent, stopLive(ownership)), ownership) + }) + const ownedByScope = ( start: ( - ownership: Pick + hooks: OwnershipHooks ) => Effect.Effect, StartError, Requirements> ): Effect.Effect, StartError, Requirements | Scope.Scope> => Effect.flatMap(Effect.scope, (parent) => - Effect.suspend(() => { - let owner: Scope.Closeable | undefined - let terminated = false - const onStopSync = () => { - terminated = true - if (owner !== undefined) Scope.closeUnsafe(owner, Exit.void) - } - return Effect.uninterruptible( - Effect.flatMap(start({ onStopSync }), (ref) => { - if (terminated) return Effect.succeed(ref) - owner = Scope.forkUnsafe(parent) - return Effect.as(Scope.addFinalizer(owner, ref.stop), ref) - }) - ) - })) + Effect.flatMap(ownershipOf(parent), (ownership) => + Effect.flatMap( + start(ownership.hooks), + (ref) => ownership.closed ? Effect.as(ref.stop, ref) : Effect.succeed(ref) + ))) const startProcessWithStrategy = Effect.fnUntraced(function*( logic: ProcessLogic, @@ -77,15 +108,15 @@ const startProcessWithStrategy = Effect.fnUntraced(function*( runtime } if (strategy === "generic") { - return yield* ownedByScope((ownership) => startGenericInternal(logic, { ...internalOptions, ...ownership })) + return yield* ownedByScope((hooks) => startGenericInternal(logic, { ...internalOptions, ...hooks })) } if (strategy === "compiled") { if (logic.execution?._tag !== "Compiled") { return yield* Effect.die(new Error("Machine cannot force the compiled runtime for generic process logic")) } - return yield* ownedByScope((ownership) => startCompactCompiledInternal(logic, { ...internalOptions, ...ownership })) + return yield* ownedByScope((hooks) => startCompactCompiledInternal(logic, { ...internalOptions, ...hooks })) } - return yield* ownedByScope((ownership) => startLogicInternal(logic, { ...internalOptions, ...ownership })) + return yield* ownedByScope((hooks) => startLogicInternal(logic, { ...internalOptions, ...hooks })) }) /** @internal Test-only startup strategy selection. */ @@ -131,20 +162,20 @@ export const startProcess: < } ) { const runtime = yield* makeRuntime - return yield* ownedByScope((ownership) => + return yield* ownedByScope((hooks) => startLogicInternal( logic, options === undefined ? { detached: true, runtime, - ...ownership + ...hooks } : { ...options, detached: true, runtime, - ...ownership + ...hooks } ) ) @@ -189,8 +220,8 @@ const prepareProcessWithStrategy = Effect.fnUntraced(function*< inspectionRoot: true, origin: { _tag: "Root" } } - const initialize = (ownership: Pick) => { - const startOptions = { ...internalOptions, ...ownership } + const initialize = (hooks: OwnershipHooks) => { + const startOptions = { ...internalOptions, ...hooks } return strategy === "generic" ? startGenericInternal(logic, startOptions) : strategy === "compiled" diff --git a/packages/effect-machine/src/internal/machine/runtimeCompiled.ts b/packages/effect-machine/src/internal/machine/runtimeCompiled.ts index 327e5f34..cd230923 100644 --- a/packages/effect-machine/src/internal/machine/runtimeCompiled.ts +++ b/packages/effect-machine/src/internal/machine/runtimeCompiled.ts @@ -648,7 +648,7 @@ class CompiledProcess implements MachineRef { if (this.compiledContext !== undefined) { this.compiledContext.executionState = undefined } - this.options.onStopSync?.() + this.options.onStopSync?.(this) this.completion = CompiledStoppedCompletion this.lifecycle = "Completed" if (this.waiter !== undefined) { @@ -816,7 +816,7 @@ class CompiledProcess implements MachineRef { : Effect.void ), Effect.andThen(Effect.sync(() => { - this.options.onStopSync?.() + this.options.onStopSync?.(this) this.runState = "Idle" this.worker = undefined this.interruptRequested = false diff --git a/packages/effect-machine/src/internal/machine/runtimeGeneric.ts b/packages/effect-machine/src/internal/machine/runtimeGeneric.ts index b5a5c550..1e4a534d 100644 --- a/packages/effect-machine/src/internal/machine/runtimeGeneric.ts +++ b/packages/effect-machine/src/internal/machine/runtimeGeneric.ts @@ -192,7 +192,7 @@ export const startGenericInternal: StartProcess = Effect.fnUntraced( Effect.andThen(options.inspectionRoot === true && inspector !== undefined ? inspector.close : Effect.void) ) } - const cleanup = onStopSync === undefined ? onStop ?? Effect.void : Effect.sync(onStopSync) + const cleanup = onStopSync === undefined ? onStop ?? Effect.void : Effect.sync(() => onStopSync(ref)) const currentCausation = inspector === undefined ? noCausation : (): Inspection.Causation | undefined => messageCausation(inFlightMessage, initializing) diff --git a/packages/effect-machine/src/internal/machine/runtimeProtocol.ts b/packages/effect-machine/src/internal/machine/runtimeProtocol.ts index 6585b4e3..f4cfac18 100644 --- a/packages/effect-machine/src/internal/machine/runtimeProtocol.ts +++ b/packages/effect-machine/src/internal/machine/runtimeProtocol.ts @@ -628,7 +628,8 @@ export interface StartInternalOptions { ) => Effect.Effect readonly onReadySync?: (ref: MachineRef) => boolean readonly onStop?: Effect.Effect - readonly onStopSync?: () => void + /** Runs synchronously once the process has terminated, with its reference. */ + readonly onStopSync?: (ref: MachineRef) => void readonly skipStoppedOutcome?: boolean readonly parent?: ProcessAddress readonly runtime: ProcessRuntime diff --git a/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts b/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts index 9c4c5a13..4a1e37e6 100644 --- a/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts +++ b/packages/effect-machine/test/internal/machine/ScopedOwnership.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest" -import { Effect, Exit, Scope } from "effect" +import { Effect, Exit, Scope, Stream } from "effect" import { Machine } from "../../../src/index.js" const Events = Machine.events({ Finish: {} }) @@ -19,13 +19,13 @@ const finalizerCount = (scope: Scope.Closeable): number => { } describe("Machine scoped ownership", () => { - it.effect("terminated machines leave the owning scope", () => + it.effect("an owning scope holds one finalizer and releases terminated machines", () => Effect.gen(function*() { const owner = yield* Scope.make() const stopped = yield* Machine.start(machine).pipe(Scope.provide(owner)) const completed = yield* Machine.start(machine).pipe(Scope.provide(owner)) const running = yield* Machine.start(machine).pipe(Scope.provide(owner)) - assert.strictEqual(finalizerCount(owner), 3) + assert.strictEqual(finalizerCount(owner), 1) yield* stopped.stop yield* completed.send(Events.Finish()) yield* completed.join @@ -35,11 +35,31 @@ describe("Machine scoped ownership", () => { assert.strictEqual(finalizerCount(owner), 0) })) + it.effect("closing the owning scope stops live machines in reverse start order", () => + Effect.gen(function*() { + const owner = yield* Scope.make() + const order: Array = [] + const refs = yield* Effect.forEach(["first", "second", "third"], (name) => + Effect.gen(function*() { + const ref = yield* Machine.start(machine).pipe(Scope.provide(owner)) + yield* Machine.watch(ref).pipe( + Stream.runForEach(() => Effect.sync(() => order.push(name))), + Effect.forkChild({ startImmediately: true }) + ) + return ref + })) + yield* Scope.close(owner, Exit.void) + for (const ref of refs) assert.strictEqual((yield* ref.snapshot).status, "stopped") + assert.deepStrictEqual(order, ["third", "second", "first"]) + })) + it.effect("starting inside a closed scope stops the machine immediately", () => Effect.gen(function*() { const owner = yield* Scope.make() yield* Scope.close(owner, Exit.void) const ref = yield* Machine.start(machine).pipe(Scope.provide(owner)) assert.strictEqual((yield* ref.snapshot).status, "stopped") + const again = yield* Machine.start(machine).pipe(Scope.provide(owner)) + assert.strictEqual((yield* again.snapshot).status, "stopped") })) }) From c81c31014f5aeaeb7fd8fb9ca78f71e1588aaeda Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 10:08:16 +0000 Subject: [PATCH 5/5] Decline readiness for machines started in a closed scope The ownership record's ready hook refuses a machine once its scope closed, so the runtime stops it itself and the start path needs no extra step. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MHnF2EL6ReDTKwoM9TVH7L --- .../src/internal/machine/runtime.ts | 56 +++++++++---------- 1 file changed, 28 insertions(+), 28 deletions(-) diff --git a/packages/effect-machine/src/internal/machine/runtime.ts b/packages/effect-machine/src/internal/machine/runtime.ts index aba858e7..afa043f8 100644 --- a/packages/effect-machine/src/internal/machine/runtime.ts +++ b/packages/effect-machine/src/internal/machine/runtime.ts @@ -53,41 +53,41 @@ const stopLive = (ownership: ScopeOwnership): Effect.Effect => return Effect.forEach(refs, (ref) => ref.stop, { discard: true }) }) -const ownershipOf = (parent: Scope.Scope): Effect.Effect => - Effect.suspend(() => { - const existing = ownerships.get(parent) - if (existing !== undefined) return Effect.succeed(existing) - const live = new Set>() - const ownership: ScopeOwnership = { - live, - hooks: { - onReadySync: (ref) => { - live.add(ref) - return true - }, - onStopSync: (ref) => { - live.delete(ref) - } +const makeOwnership = (): ScopeOwnership => { + const live = new Set>() + const ownership: ScopeOwnership = { + live, + hooks: { + // Declining readiness makes the runtime stop the machine itself, which + // covers a Scope that closed before or while the machine started. + onReadySync: (ref) => { + if (ownership.closed) return false + live.add(ref) + return true }, - closed: false - } - ownerships.set(parent, ownership) - // A closed Scope runs the finalizer immediately, which marks the record - // closed so machines started afterwards stop right away. - return Effect.as(Scope.addFinalizer(parent, stopLive(ownership)), ownership) - }) + onStopSync: (ref) => { + live.delete(ref) + } + }, + closed: false + } + return ownership +} const ownedByScope = ( start: ( hooks: OwnershipHooks ) => Effect.Effect, StartError, Requirements> ): Effect.Effect, StartError, Requirements | Scope.Scope> => - Effect.flatMap(Effect.scope, (parent) => - Effect.flatMap(ownershipOf(parent), (ownership) => - Effect.flatMap( - start(ownership.hooks), - (ref) => ownership.closed ? Effect.as(ref.stop, ref) : Effect.succeed(ref) - ))) + Effect.flatMap(Effect.scope, (parent) => { + const existing = ownerships.get(parent) + if (existing !== undefined) return start(existing.hooks) + const ownership = makeOwnership() + ownerships.set(parent, ownership) + // A closed Scope runs the finalizer immediately, which marks the record + // closed so the machine started next stops right away. + return Effect.andThen(Scope.addFinalizer(parent, stopLive(ownership)), start(ownership.hooks)) + }) const startProcessWithStrategy = Effect.fnUntraced(function*( logic: ProcessLogic,