diff --git a/.changeset/direct-invoked-snapshots.md b/.changeset/direct-invoked-snapshots.md new file mode 100644 index 0000000..36afa18 --- /dev/null +++ b/.changeset/direct-invoked-snapshots.md @@ -0,0 +1,5 @@ +--- +"@typeonce/effect-machine": patch +--- + +Deliver invoked child snapshots directly from the child runtime, removing the replay PubSub and watcher fiber previously retained by every snapshot-mapped invocation. diff --git a/perf/runtime/README.md b/perf/runtime/README.md index 92b5910..25543fe 100644 --- a/perf/runtime/README.md +++ b/perf/runtime/README.md @@ -58,10 +58,12 @@ The fitted heap slope is the primary idle-capacity metric. Compare adjacent profiles to attribute retained memory: raw process to idle statechart isolates statechart machinery, two independent machines to parent-with-child isolates relationship bookkeeping, while the two observed parent-child profiles isolate -registry and invoked-snapshot observation. The Effect profiles are primitive -lower bounds, not feature-equivalent competitors. Resident memory is reported -as a raw diagnostic because V8 and the operating-system allocator can reuse -already committed pages. The +registry and invoked-snapshot observation. Invoked snapshot mapping uses a +direct, state-scoped delivery path; its profile measures the retained callback +and mapping state rather than a general `changes` stream subscription. The +Effect profiles are primitive lower bounds, not feature-equivalent competitors. +Resident memory is reported as a raw diagnostic because V8 and the +operating-system allocator can reuse already committed pages. The capacity-per-GiB value is a linear estimate that excludes shared process overhead; it is not a run-until-OOM limit. diff --git a/perf/runtime/counter.mjs b/perf/runtime/counter.mjs index 192ea48..be0b15e 100644 --- a/perf/runtime/counter.mjs +++ b/perf/runtime/counter.mjs @@ -357,7 +357,7 @@ export const effectMachineAdapter = { stop: stopObservedChildCounters }, "snapshot-observed-parent-with-child": { - label: "Parent with invoked child snapshot watcher", + label: "Parent with observed invoked child snapshots", start: startSnapshotObservedChildCounters, stop: stopChildCounters } diff --git a/src/internal/machineProcess.ts b/src/internal/machineProcess.ts index 9153301..9e964be 100644 --- a/src/internal/machineProcess.ts +++ b/src/internal/machineProcess.ts @@ -6,13 +6,11 @@ import * as Effect from "effect/Effect" import * as Exit from "effect/Exit" -import * as Fiber from "effect/Fiber" import * as HashMap from "effect/HashMap" import * as Option from "effect/Option" import * as Queue from "effect/Queue" import * as Ref from "effect/Ref" import type * as Schema from "effect/Schema" -import * as Stream from "effect/Stream" import type { ActionError, ExecutionServices, Machine, Runtime } from "../Machine.js" import { ChildAlreadyExistsError, InfiniteTransitionError, MachineSchemaDecodeError } from "./machineErrors.js" import type { StartupError, StoppedError } from "./machineErrors.js" @@ -34,7 +32,6 @@ type AnyInvokeConfig = Machine.InvokeConfig | undefined readonly childId: string readonly path: string } @@ -240,14 +237,7 @@ const makeProcessLogic: < return Option.isSome(current) && current.value.token === token }) ) - const stopInvokeSession = (session: InvokeSession): Effect.Effect => - Effect.all( - [ - ...(session.watcher === undefined ? [] : [Fiber.interrupt(session.watcher)]), - context.stopChild(session.childId) - ], - { discard: true, concurrency: "unbounded" } - ) + const stopInvokeSession = (session: InvokeSession): Effect.Effect => context.stopChild(session.childId) const removeInvoke = ( key: string, token: symbol | undefined @@ -304,49 +294,25 @@ const makeProcessLogic: < }) ) } - const startInvokeSnapshotWatcher = Effect.fnUntraced(function*( + const handleInvokeSnapshot = ( config: AnyInvokeConfig, - child: internalRuntime.MachineRef, key: string, - token: symbol - ) { - if (config.snapshot === undefined) { - return - } - const mapSnapshot = config.snapshot - const watcher = yield* child.changes.pipe( - Stream.filter((snapshot) => snapshot.status === "active"), - Stream.runForEach((snapshot) => - isCurrentInvoke(key, token).pipe( - Effect.flatMap((isCurrent) => { - if (!isCurrent) { - return Effect.void - } - const mappedEvent = mapSnapshot({ id: config.id, snapshot }) - return mappedEvent === undefined - ? Effect.void - : context.self.send(mappedEvent as Machine.EventOf).pipe( - Effect.catchTag("StoppedError", () => Effect.void) - ) - }) - ) - ), - Effect.forkChild + token: symbol, + snapshot: Extract, { readonly status: "active" }> + ): Effect.Effect => + isCurrentInvoke(key, token).pipe( + Effect.flatMap((isCurrent) => { + if (!isCurrent || config.snapshot === undefined) { + return Effect.void + } + const mappedEvent = config.snapshot({ id: config.id, snapshot }) + return mappedEvent === undefined + ? Effect.void + : context.self.send(mappedEvent as Machine.EventOf).pipe( + Effect.catchTag("StoppedError", () => Effect.void) + ) + }) ) - const installed = yield* Ref.modify(invokeSessions, (sessions) => { - const current = HashMap.get(sessions, key) - if (Option.isNone(current) || current.value.token !== token) { - return [false, sessions] as const - } - return [ - true, - HashMap.set(sessions, key, { ...current.value, watcher }) - ] as const - }) - if (!installed) { - yield* Fiber.interrupt(watcher) - } - }) const startInvoke = Effect.fnUntraced(function*>( path: StateId, config: AnyInvokeConfig @@ -358,7 +324,7 @@ const makeProcessLogic: < const reserved = yield* Ref.modify(invokeSessions, (sessions) => HashMap.has(sessions, key) ? [false, sessions] as const - : [true, HashMap.set(sessions, key, { token, watcher: undefined, childId, path })] as const) + : [true, HashMap.set(sessions, key, { token, childId, path })] as const) if (!reserved) { return yield* Effect.fail(new ChildAlreadyExistsError({ id: invokeId })) } @@ -369,7 +335,7 @@ const makeProcessLogic: < isCurrent ? context.self.send(event as Machine.EventOf) : Effect.void ) ) - const child = yield* context.spawn( + yield* context.spawn( { initial: (childScope) => logic.initial({ ...childScope, sendParent }), run: (childContext) => logic.run({ ...childContext, sendParent }) @@ -377,7 +343,11 @@ const makeProcessLogic: < { id: childId, ...(config.descriptor === undefined ? undefined : { descriptor: config.descriptor }), - onOutcome: (outcome) => handleInvokeOutcome(config, key, token, outcome) + onOutcome: (outcome) => handleInvokeOutcome(config, key, token, outcome), + ...(config.snapshot === undefined ? undefined : { + [internalRuntime.activeSnapshotObserver]: (snapshot) => + handleInvokeSnapshot(config, key, token, snapshot) + }) } ).pipe( Effect.onExit((exit) => @@ -386,7 +356,6 @@ const makeProcessLogic: < : Effect.void ) ) - yield* startInvokeSnapshotWatcher(config, child, key, token) }) const startInvokes: ( configuration: Model.ActiveConfiguration, diff --git a/src/internal/machineRuntime.ts b/src/internal/machineRuntime.ts index 69c834b..a4f0624 100644 --- a/src/internal/machineRuntime.ts +++ b/src/internal/machineRuntime.ts @@ -43,6 +43,9 @@ type ChildEntry = type ChildSelector = string | ChildDescriptor type ChildKey = string | symbol +/** @internal */ +export const activeSnapshotObserver: unique symbol = Symbol.for("effect/Machine/activeSnapshotObserver") + interface ChildRegistrySnapshot { readonly closed: boolean readonly revision: number @@ -183,6 +186,9 @@ export interface ProcessSpawn { readonly onOutcome?: ( outcome: RuntimeOutcome ) => Effect.Effect + readonly [activeSnapshotObserver]?: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect } ): Effect.Effect< MachineRef, @@ -257,6 +263,17 @@ const classifyOutcome = ( } } +const notifyActiveSnapshot = ( + onSnapshot: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect, + snapshot: Extract, { readonly status: "active" }> +): Effect.Effect => + Effect.suspend(() => onSnapshot(snapshot)).pipe( + Effect.exit, + Effect.asVoid + ) + export const watch = ( ref: MachineRef ): Stream.Stream> => @@ -281,6 +298,9 @@ interface StartInternalOptions { readonly detached?: boolean readonly id?: string readonly onOutcome?: (outcome: RuntimeOutcome) => Effect.Effect + readonly onSnapshot?: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect readonly onReady?: ( ref: MachineRef, requestStop: Effect.Effect @@ -567,6 +587,9 @@ const startInternal: < readonly onOutcome?: ( outcome: RuntimeOutcome ) => Effect.Effect + readonly [activeSnapshotObserver]?: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect } ): Effect.Effect< MachineRef, @@ -581,6 +604,9 @@ const startInternal: < readonly onOutcome?: ( outcome: RuntimeOutcome ) => Effect.Effect + readonly [activeSnapshotObserver]?: ( + snapshot: Extract, { readonly status: "active" }> + ) => Effect.Effect } ): Effect.Effect< MachineRef, @@ -603,6 +629,9 @@ const startInternal: < detached: true, ...(spawnOptions?.id === undefined ? undefined : { id: spawnOptions.id }), ...(spawnOptions?.onOutcome === undefined ? undefined : { onOutcome: spawnOptions.onOutcome }), + ...(spawnOptions?.[activeSnapshotObserver] === undefined + ? undefined + : { onSnapshot: spawnOptions[activeSnapshotObserver] }), onReady: (child, requestChildStop) => Effect.sync(() => { startedChild = child @@ -681,12 +710,22 @@ const startInternal: < state: initial } }) - const publishSnapshot = ( + const publishSnapshot: ( snapshot: VersionedSnapshot - ): Effect.Effect> => - snapshot.changes === undefined - ? Effect.succeed(snapshot) - : PubSub.publish(snapshot.changes, [snapshot] as const).pipe(Effect.as(snapshot)) + ) => Effect.Effect> = options.onSnapshot === undefined + ? (snapshot) => + snapshot.changes === undefined + ? Effect.succeed(snapshot) + : PubSub.publish(snapshot.changes, [snapshot] as const).pipe(Effect.as(snapshot)) + : (snapshot) => { + const publish = snapshot.changes === undefined + ? Effect.succeed(snapshot) + : PubSub.publish(snapshot.changes, [snapshot] as const).pipe(Effect.as(snapshot)) + const runtimeSnapshot = snapshot.snapshot + return runtimeSnapshot.status !== "active" + ? publish + : publish.pipe(Effect.tap(() => notifyActiveSnapshot(options.onSnapshot!, runtimeSnapshot))) + } const completeChanges = ( snapshot: VersionedSnapshot @@ -945,6 +984,9 @@ const startInternal: < if (options.onReady !== undefined) { yield* options.onReady(ref, requestStop) } + if (options.onSnapshot !== undefined) { + yield* notifyActiveSnapshot(options.onSnapshot, { status: "active", state: initial }) + } const reserveTermination = (termination: ProcessTermination) => { switch (termination._tag) { diff --git a/test/MachineProcessLifecycle.test.ts b/test/MachineProcessLifecycle.test.ts index ff3dddb..a864c3f 100644 --- a/test/MachineProcessLifecycle.test.ts +++ b/test/MachineProcessLifecycle.test.ts @@ -433,6 +433,60 @@ describe("machine process lifecycle", () => { yield* parent.stop })) + it.effect("delivers committed active child snapshots directly and in order", () => + Effect.gen(function*() { + const parentScope = yield* Deferred.make>() + const snapshots = yield* Ref.make>([]) + const parent = yield* MachineRuntime.startProcess({ + initial: (scope) => Deferred.succeed(parentScope, scope).pipe(Effect.as(undefined)), + run: () => Effect.never + }) + const child = yield* (yield* Deferred.await(parentScope)).spawn( + Machine.logic({ + initial: 0, + run: ({ setState }) => + setState(1).pipe( + Effect.andThen(setState(2)), + Effect.as("output") + ) + }), + { + id: "child", + [MachineRuntime.activeSnapshotObserver]: (snapshot) => + Ref.update(snapshots, (current) => [...current, snapshot.state]) + } + ) + + assert.strictEqual(yield* child.join, "output") + assert.deepStrictEqual(yield* Ref.get(snapshots), [0, 1, 2]) + yield* parent.stop + })) + + it.effect("isolates active child state updates from snapshot callback defects", () => + Effect.gen(function*() { + const parentScope = yield* Deferred.make>() + const parent = yield* MachineRuntime.startProcess({ + initial: (scope) => Deferred.succeed(parentScope, scope).pipe(Effect.as(undefined)), + run: () => Effect.never + }) + const child = yield* (yield* Deferred.await(parentScope)).spawn( + Machine.logic({ + initial: 0, + run: ({ setState }) => setState(1).pipe(Effect.as("output")) + }), + { id: "child", [MachineRuntime.activeSnapshotObserver]: () => Effect.die("callback defect") } + ) + + assert.strictEqual(yield* child.join, "output") + assert.deepStrictEqual(yield* child.snapshot, { + status: "done", + state: 1, + output: "output" + }) + assert.deepStrictEqual(yield* parent.snapshot, { status: "active", state: undefined }) + yield* parent.stop + })) + it.effect("publishes and cleans up exactly once when stop races process completion", () => Effect.gen(function*() { const cleanupCount = yield* Ref.make(0)