Deliver invoked snapshots directly

.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.

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.
 

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
     }

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<any, any, any, any, any, any, any, a
 
 interface InvokeSession {
   readonly token: symbol
-  readonly watcher: Fiber.Fiber<void> | 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<void> =>
-            Effect.all(
-              [
-                ...(session.watcher === undefined ? [] : [Fiber.interrupt(session.watcher)]),
-                context.stopChild(session.childId)
-              ],
-              { discard: true, concurrency: "unbounded" }
-            )
+          const stopInvokeSession = (session: InvokeSession): Effect.Effect<void> => 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<any, any, any, any>,
             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<Events>).pipe(
-                        Effect.catchTag("StoppedError", () => Effect.void)
-                      )
-                  })
-                )
-              ),
-              Effect.forkChild
+            token: symbol,
+            snapshot: Extract<internalRuntime.RuntimeSnapshot<any, any, any>, { readonly status: "active" }>
+          ): Effect.Effect<void> =>
+            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<Events>).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*<StateId extends Machine.StateIdentifier<States>>(
             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,15 +335,19 @@ const makeProcessLogic: <
                   isCurrent ? context.self.send(event as Machine.EventOf<Events>) : Effect.void
                 )
               )
-            const child = yield* context.spawn(
+            yield* context.spawn(
               {
                 initial: (childScope) => logic.initial({ ...childScope, sendParent }),
                 run: (childContext) => logic.run({ ...childContext, sendParent })
               },
               {
                 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,

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<ChildState, ChildError, ChildOutput>
       ) => Effect.Effect<void>
+      readonly [activeSnapshotObserver]?: (
+        snapshot: Extract<RuntimeSnapshot<ChildState, ChildError, ChildOutput>, { readonly status: "active" }>
+      ) => Effect.Effect<void>
     }
   ): Effect.Effect<
     MachineRef<ChildState, ChildEvent, ChildError, ChildOutput>,
@@ -257,6 +263,17 @@ const classifyOutcome = <State, Error, Output>(
   }
 }
 
+const notifyActiveSnapshot = <State, Error, Output>(
+  onSnapshot: (
+    snapshot: Extract<RuntimeSnapshot<State, Error, Output>, { readonly status: "active" }>
+  ) => Effect.Effect<void>,
+  snapshot: Extract<RuntimeSnapshot<State, Error, Output>, { readonly status: "active" }>
+): Effect.Effect<void> =>
+  Effect.suspend(() => onSnapshot(snapshot)).pipe(
+    Effect.exit,
+    Effect.asVoid
+  )
+
 export const watch = <State, Event, Error = never, Output = never>(
   ref: MachineRef<State, Event, Error, Output>
 ): Stream.Stream<RuntimeOutcome<State, Error, Output>> =>
@@ -281,6 +298,9 @@ interface StartInternalOptions {
   readonly detached?: boolean
   readonly id?: string
   readonly onOutcome?: (outcome: RuntimeOutcome<any, any, any>) => Effect.Effect<void>
+  readonly onSnapshot?: (
+    snapshot: Extract<RuntimeSnapshot<any, any, any>, { readonly status: "active" }>
+  ) => Effect.Effect<void>
   readonly onReady?: (
     ref: MachineRef<any, any, any, any>,
     requestStop: Effect.Effect<void>
@@ -567,6 +587,9 @@ const startInternal: <
       readonly onOutcome?: (
         outcome: RuntimeOutcome<ChildState, ChildError, ChildOutput>
       ) => Effect.Effect<void>
+      readonly [activeSnapshotObserver]?: (
+        snapshot: Extract<RuntimeSnapshot<ChildState, ChildError, ChildOutput>, { readonly status: "active" }>
+      ) => Effect.Effect<void>
     }
   ): Effect.Effect<
     MachineRef<ChildState, ChildEvent, ChildError, ChildOutput>,
@@ -581,6 +604,9 @@ const startInternal: <
       readonly onOutcome?: (
         outcome: RuntimeOutcome<ChildState, ChildError, ChildOutput>
       ) => Effect.Effect<void>
+      readonly [activeSnapshotObserver]?: (
+        snapshot: Extract<RuntimeSnapshot<ChildState, ChildError, ChildOutput>, { readonly status: "active" }>
+      ) => Effect.Effect<void>
     }
   ): Effect.Effect<
     MachineRef<ChildState, ChildEvent, ChildError, ChildOutput>,
@@ -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<State, Error, Output>
-  ): Effect.Effect<VersionedSnapshot<State, Error, Output>> =>
-    snapshot.changes === undefined
-      ? Effect.succeed(snapshot)
-      : PubSub.publish(snapshot.changes, [snapshot] as const).pipe(Effect.as(snapshot))
+  ) => Effect.Effect<VersionedSnapshot<State, Error, Output>> = 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<State, Error, Output>
@@ -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) {

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<MachineRuntime.ProcessScope<never>>()
+      const snapshots = yield* Ref.make<ReadonlyArray<number>>([])
+      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<MachineRuntime.ProcessScope<never>>()
+      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)