Compact invoked child supervision (#44)

* Compact invoked child supervision

* Skip stopped invoke outcome lookups

Author: SandroMaglione

.changeset/direct-invokes-report.md

@@ -0,0 +1,5 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Reduce invoked-child memory and lifecycle overhead by delivering terminal outcomes directly through process supervision and allocating a watcher fiber only for invokes that map active child snapshots.

perf/runtime/README.md

@@ -17,7 +17,8 @@ The command reports:
 - parent-with-child start-and-stop throughput;
 - heap and resident-memory growth at 100, 500, and 1,000 live units, including
   a raw managed process, an idle statechart, two independent statecharts, a
-  parent with one child, and that relationship with child observation active;
+  parent with one child, that relationship with child-registry observation
+  active, and an invoked child whose active snapshots are observed;
 - lower-bound memory profiles for Effect itself: a suspended fiber, a queue
   with a waiting fiber, and a minimal mailbox/state/completion actor shell.
 
@@ -56,10 +57,11 @@ same runtime guarantees.
 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, and unobserved to observed parent-child isolates
-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
+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
 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

@@ -71,6 +71,20 @@ const counterParentMachine = Machine.make({
   }
 })
 
+const counterSnapshotParentMachine = Machine.make({
+  id: "RuntimeBenchmarkSnapshotCounterParent",
+  states: ParentStates.states,
+  events: [],
+  initial: () => ParentStates.initial.Active(ParentState.cases.Active.make({}))
+}).handle({
+  Active: {
+    invoke: Machine.invokeMachine({
+      child: CounterChild,
+      snapshot: () => undefined
+    })
+  }
+})
+
 export const incrementEvent = CounterEvent.cases.Increment.make({})
 const finishEvent = CounterEvent.cases.Finish.make({})
 
@@ -254,6 +268,20 @@ const startObservedChildCounters = (count) =>
     )
   )
 
+const startSnapshotObservedChildCounters = (count) =>
+  Effect.runPromise(
+    Effect.forEach(
+      Array.from({ length: count }),
+      () =>
+        Effect.gen(function*() {
+          const parent = yield* Machine.start(counterSnapshotParentMachine)
+          yield* waitForCounterChild(parent)
+          return parent
+        }),
+      { concurrency: 1 }
+    )
+  )
+
 const stopObservedChildCounters = (units) =>
   Effect.runPromise(
     Effect.forEach(
@@ -327,6 +355,11 @@ export const effectMachineAdapter = {
       label: "Parent with observed child registry",
       start: startObservedChildCounters,
       stop: stopObservedChildCounters
+    },
+    "snapshot-observed-parent-with-child": {
+      label: "Parent with invoked child snapshot watcher",
+      start: startSnapshotObservedChildCounters,
+      stop: stopChildCounters
     }
   }
 }

src/internal/machineProcess.ts

@@ -6,12 +6,12 @@
 
 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 Scope from "effect/Scope"
 import * as Stream from "effect/Stream"
 import type { ActionError, ExecutionServices, Machine, Runtime } from "../Machine.js"
 import { ChildAlreadyExistsError, InfiniteTransitionError, MachineSchemaDecodeError } from "./machineErrors.js"
@@ -34,7 +34,7 @@ type AnyInvokeConfig = Machine.InvokeConfig<any, any, any, any, any, any, any, a
 
 interface InvokeSession {
   readonly token: symbol
-  readonly scope: Scope.Closeable
+  readonly watcher: Fiber.Fiber<void> | undefined
   readonly childId: string
   readonly path: string
 }
@@ -240,10 +240,17 @@ 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 removeInvoke = (
             key: string,
-            token: symbol | undefined,
-            exit: Exit.Exit<unknown, unknown>
+            token: symbol | undefined
           ): Effect.Effect<void> =>
             Ref.modify(invokeSessions, (sessions) => {
               const current = HashMap.get(sessions, key)
@@ -254,80 +261,91 @@ const makeProcessLogic: <
               Effect.flatMap((session) =>
                 session === undefined
                   ? Effect.void
-                  : Scope.close(session.scope, exit).pipe(
-                    Effect.andThen(context.stopChild(session.childId))
-                  )
+                  : stopInvokeSession(session)
               )
             )
-          const stopInvoke = (key: string, exit: Exit.Exit<unknown, unknown>): Effect.Effect<void> =>
-            removeInvoke(key, undefined, exit)
-          const stopAllInvokes = (exit: Exit.Exit<unknown, unknown>): Effect.Effect<void> =>
-            Ref.modify(invokeSessions, (sessions) => [HashMap.toEntries(sessions), HashMap.empty()] as const).pipe(
+          const stopInvoke = (key: string): Effect.Effect<void> => removeInvoke(key, undefined)
+          const stopAllInvokes: Effect.Effect<void> = Ref.modify(invokeSessions, (sessions) =>
+            [HashMap.toEntries(sessions), HashMap.empty()] as const).pipe(
               Effect.flatMap((sessions) =>
                 Effect.all(
                   sessions.map(([, session]) =>
-                    Scope.close(session.scope, exit).pipe(
-                      Effect.andThen(context.stopChild(session.childId))
-                    )
+                    stopInvokeSession(session)
                   ),
                   { discard: true, concurrency: "unbounded" }
                 )
               )
             )
-          const startInvokeWatchers = Effect.fnUntraced(function*(
+          const handleInvokeOutcome = (
             config: AnyInvokeConfig,
-            child: internalRuntime.MachineRef<any, any, any, any>,
             key: string,
             token: symbol,
-            scope: Scope.Closeable
+            outcome: internalRuntime.RuntimeOutcome<any, any, any>
+          ): Effect.Effect<void> => {
+            if (outcome._tag === "Stopped") {
+              return Effect.void
+            }
+            return isCurrentInvoke(key, token).pipe(
+              Effect.flatMap((isCurrent) => {
+                if (!isCurrent) {
+                  return Effect.void
+                }
+                if (outcome._tag === "Done") {
+                  const mappedEvent = config.onDone === undefined
+                    ? outcome.output
+                    : config.onDone({ id: config.id, output: outcome.output })
+                  return mappedEvent === undefined
+                    ? Effect.void
+                    : context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
+                      Effect.catchTag("StoppedError", () => Effect.void)
+                    )
+                }
+                return context.failCause(outcome.cause)
+              })
+            )
+          }
+          const startInvokeSnapshotWatcher = Effect.fnUntraced(function*(
+            config: AnyInvokeConfig,
+            child: internalRuntime.MachineRef<any, any, any, any>,
+            key: string,
+            token: symbol
           ) {
-            if (config.snapshot !== undefined) {
-              const mapSnapshot = config.snapshot
-              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.forkIn(scope),
-                Effect.asVoid
-              )
+            if (config.snapshot === undefined) {
+              return
             }
-            yield* internalRuntime.watch(child).pipe(
-              Stream.runForEach((outcome) =>
+            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 || outcome._tag === "Stopped") {
+                    if (!isCurrent) {
                       return Effect.void
                     }
-                    if (outcome._tag === "Done") {
-                      const mappedEvent = config.onDone === undefined
-                        ? outcome.output
-                        : config.onDone({ id: config.id, output: outcome.output })
-                      return mappedEvent === undefined
-                        ? Effect.void
-                        : context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
-                          Effect.catchTag("StoppedError", () => Effect.void)
-                        )
-                    }
-                    return context.failCause(outcome.cause)
+                    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.forkIn(scope),
-              Effect.asVoid
+              Effect.forkChild
             )
+            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,
@@ -337,13 +355,11 @@ const makeProcessLogic: <
             const invokeId = String(config.id)
             const key = makeInvokeSessionKey(path, invokeId)
             const childId = config.address === undefined ? makeInvokeChildId(path, invokeId) : String(config.address)
-            const scope = yield* Scope.make("parallel")
             const reserved = yield* Ref.modify(invokeSessions, (sessions) =>
               HashMap.has(sessions, key)
                 ? [false, sessions] as const
-                : [true, HashMap.set(sessions, key, { token, scope, childId, path })] as const)
+                : [true, HashMap.set(sessions, key, { token, watcher: undefined, childId, path })] as const)
             if (!reserved) {
-              yield* Scope.close(scope, Exit.void)
               return yield* Effect.fail(new ChildAlreadyExistsError({ id: invokeId }))
             }
             const logic = config.src()
@@ -358,24 +374,19 @@ const makeProcessLogic: <
                 initial: (childScope) => logic.initial({ ...childScope, sendParent }),
                 run: (childContext) => logic.run({ ...childContext, sendParent })
               },
-              config.descriptor === undefined
-                ? { id: childId }
-                : { id: childId, descriptor: config.descriptor }
+              {
+                id: childId,
+                ...(config.descriptor === undefined ? undefined : { descriptor: config.descriptor }),
+                onOutcome: (outcome) => handleInvokeOutcome(config, key, token, outcome)
+              }
             ).pipe(
               Effect.onExit((exit) =>
                 Exit.isFailure(exit)
-                  ? Ref.update(invokeSessions, (sessions) => {
-                    const current = HashMap.get(sessions, key)
-                    return Option.isSome(current) && current.value.token === token
-                      ? HashMap.remove(sessions, key)
-                      : sessions
-                  }).pipe(
-                    Effect.andThen(Scope.close(scope, Exit.failCause(exit.cause)))
-                  )
+                  ? removeInvoke(key, token)
                   : Effect.void
               )
             )
-            yield* startInvokeWatchers(config, child, key, token, scope)
+            yield* startInvokeSnapshotWatcher(config, child, key, token)
           })
           const startInvokes: (
             configuration: Model.ActiveConfiguration,
@@ -412,7 +423,7 @@ const makeProcessLogic: <
                   internalPlanner.sortExitPaths(machine, paths).flatMap((path) =>
                     HashMap.toEntries(sessions)
                       .filter(([, session]) => session.path === path)
-                      .map(([key]) => stopInvoke(key, Exit.void))
+                      .map(([key]) => stopInvoke(key))
                   ),
                   { discard: true, concurrency: "unbounded" }
                 )
@@ -481,7 +492,7 @@ const makeProcessLogic: <
 
                     if (planned.done) {
                       terminal = { output: planned.output as Output }
-                      yield* stopAllInvokes(Exit.succeed(planned.output))
+                      yield* stopAllInvokes
                     } else {
                       if (changed) {
                         for (const [path, entryEvent] of entryEvents) {
@@ -509,7 +520,7 @@ const makeProcessLogic: <
             }
             return terminal.output
           }).pipe(
-            Effect.onExit((exit) => stopAllInvokes(exit))
+            Effect.onExit(() => stopAllInvokes)
           )
         }),
         context