Strengthen machine strategy guardrails (#76)

Author: SandroMaglione

.changeset/strong-machines-compare.md

@@ -0,0 +1,5 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Add direct generic/indexed planner and generic/compiled runtime strategy guardrails, including startup, targetless, reentry, invoke, snapshot-stability, and generated-model coverage. Expand the runtime benchmark suite with hierarchical and parallel planning, observed hierarchical execution, and generic process lifecycle measurements.

perf/runtime/counter.mjs

@@ -1,4 +1,4 @@
-import { readFileSync } from "node:fs"
+import { existsSync, readFileSync } from "node:fs"
 import { createRequire } from "node:module"
 import { dirname, join, resolve } from "node:path"
 import { fileURLToPath, pathToFileURL } from "node:url"
@@ -11,9 +11,14 @@ const effectPackagePath = implementationRequire.resolve("effect/package.json")
 const effectPackage = JSON.parse(readFileSync(effectPackagePath, "utf8"))
 const effect = await import(pathToFileURL(resolve(dirname(effectPackagePath), effectPackage.exports["."])).href)
 const { Machine } = await import(pathToFileURL(join(implementationRoot, "dist/index.js")).href)
-const machineRuntime = await import(
-  pathToFileURL(join(implementationRoot, "dist/internal/machineRuntime.js")).href
-)
+const machineRuntimePath = [
+  join(implementationRoot, "dist/internal/machine/runtime.js"),
+  join(implementationRoot, "dist/internal/machineRuntime.js")
+].find(existsSync)
+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 CounterState = Schema.TaggedUnion({
@@ -221,6 +226,12 @@ const parallelFinishEvent = makeEvent(parallelCounterMachine, HierarchicalEvent.
 export const initialCounterSnapshot = Effect.runSync(
   Machine.planInitial(counterMachine).pipe(Effect.map((planned) => planned.state))
 )
+const initialHierarchicalSnapshot = Effect.runSync(
+  Machine.planInitial(hierarchicalCounterMachine).pipe(Effect.map((planned) => planned.state))
+)
+const initialParallelSnapshot = Effect.runSync(
+  Machine.planInitial(parallelCounterMachine).pipe(Effect.map((planned) => planned.state))
+)
 
 export const counterValue = (snapshot) => snapshot.value.value
 
@@ -236,6 +247,32 @@ export const planCounterBatch = (size) => {
   )
 }
 
+const planHierarchicalCounterBatch = (size) =>
+  Effect.runSync(
+    Effect.gen(function*() {
+      let snapshot = initialHierarchicalSnapshot
+      for (let index = 0; index < size; index += 1) {
+        snapshot = (yield* Machine.plan(hierarchicalCounterMachine, snapshot, hierarchicalIncrementEvent)).next
+      }
+      return snapshot.state.value.value
+    })
+  )
+
+const planParallelCounterBatch = (size) =>
+  Effect.runSync(
+    Effect.gen(function*() {
+      let snapshot = initialParallelSnapshot
+      for (let index = 0; index < size; index += 1) {
+        snapshot = (yield* Machine.plan(
+          parallelCounterMachine,
+          snapshot,
+          index % 2 === 0 ? parallelIncrementLeftEvent : parallelIncrementRightEvent
+        )).next
+      }
+      return snapshot.states.Left.value.value + snapshot.states.Right.value.value
+    })
+  )
+
 export const startCounter = () => Effect.runPromise(Machine.start(counterMachine))
 
 export const stopCounter = (ref) => Effect.runPromise(ref.stop)
@@ -324,6 +361,16 @@ export const runCounterBurst = (ref, size) =>
 const startHierarchicalCounter = () => Effect.runPromise(Machine.start(hierarchicalCounterMachine))
 const startParallelCounter = () => Effect.runPromise(Machine.start(parallelCounterMachine))
 
+const startObservedHierarchicalCounter = () =>
+  Effect.runPromise(
+    Effect.gen(function*() {
+      const ref = yield* Machine.start(hierarchicalCounterMachine)
+      const observer = yield* ref.changes.pipe(Stream.runDrain, Effect.forkDetach)
+      yield* Effect.yieldNow
+      return { ref, observer }
+    })
+  )
+
 const runHierarchicalCounterBurst = (ref, size) =>
   Effect.runPromise(
     Effect.gen(function*() {
@@ -346,6 +393,19 @@ const runParallelCounterBurst = (ref, size) =>
     })
   )
 
+const runObservedHierarchicalCounterBurst = ({ ref, observer }, size) =>
+  Effect.runPromise(
+    Effect.gen(function*() {
+      for (let index = 0; index < size; index += 1) {
+        yield* ref.send(hierarchicalIncrementEvent)
+      }
+      yield* ref.send(hierarchicalFinishEvent)
+      const value = yield* ref.join
+      yield* Fiber.join(observer)
+      return value
+    })
+  )
+
 export const startCounters = (count) =>
   Effect.runPromise(
     Effect.forEach(
@@ -502,6 +562,26 @@ export const effectMachineAdapter = {
   stopObservedCounter,
   stopCounters,
   additionalMachineBenchmarks: [
+    {
+      id: "hierarchical-plan-counter",
+      label: "Plan transitions through a compound state",
+      unit: "transitions/s",
+      operations: ({ planningBatchSize }) => planningBatchSize,
+      expected: (operations) => operations,
+      start: () => undefined,
+      run: (_, operations) => planHierarchicalCounterBatch(operations),
+      stop: () => undefined
+    },
+    {
+      id: "parallel-plan-counter",
+      label: "Plan transitions through parallel regions",
+      unit: "transitions/s",
+      operations: ({ planningBatchSize }) => planningBatchSize,
+      expected: (operations) => operations,
+      start: () => undefined,
+      run: (_, operations) => planParallelCounterBatch(operations),
+      stop: () => undefined
+    },
     {
       id: "hierarchical-runtime-burst",
       label: "Drain burst through a compound state",
@@ -521,9 +601,34 @@ export const effectMachineAdapter = {
       start: startParallelCounter,
       run: runParallelCounterBurst,
       stop: stopCounter
+    },
+    {
+      id: "observed-hierarchical-runtime-burst",
+      label: "Drain a compound-state burst with a change observer",
+      unit: "events/s",
+      operations: ({ burstBatchSize }) => burstBatchSize,
+      expected: (operations) => operations,
+      start: startObservedHierarchicalCounter,
+      run: runObservedHierarchicalCounterBurst,
+      stop: stopObservedCounter
     }
   ],
   runtimeBenchmarks: [
+    {
+      id: "generic-process-start-stop",
+      label: "Start and stop a raw generic process",
+      unit: "processes/s",
+      async: true,
+      operations: () => 1,
+      run: async () => {
+        const refs = await startRawProcesses(1)
+        try {
+          return refs.length
+        } finally {
+          await stopCounters(refs)
+        }
+      }
+    },
     {
       id: "compiled-process-start-stop",
       label: "Start and stop a raw compiled process",

src/internal/machinePlanner.ts

@@ -2724,64 +2724,102 @@ export interface CompiledExecutionPlan {
 
 const executionPlanCache = new WeakMap<Machine.Any, CompiledExecutionPlan>()
 
+const makeActiveExecutionPlan = (machine: Machine.Any): CompiledExecutionPlan => ({
+  fromConfiguration: (configuration) => configuration,
+  toConfiguration: (state) => state as ActiveConfiguration,
+  snapshot: (state) => snapshotFromConfiguration(machine, state as ActiveConfiguration),
+  plan: (state, event) => planConfiguration(machine as any, state as ActiveConfiguration, event as any)
+})
+
+const makeIndexedExecutionPlan = (
+  machine: Machine.Any,
+  indexed: IndexedExecutionDescriptor
+): CompiledExecutionPlan => ({
+  fromConfiguration: (configuration) => indexedConfigurationFromActive(indexed, configuration),
+  toConfiguration: (state) => activeConfigurationFromIndexed(indexed, state as IndexedConfiguration),
+  snapshot: (state) => snapshotFromIndexed(indexed, state as IndexedConfiguration),
+  plan: (state, event) => planIndexedConfiguration(machine, indexed, state as IndexedConfiguration, event),
+  initial: (args) => {
+    const inputArgs = machine.input === undefined
+      ? args
+      : args.length === 0
+      ? (decodeInputSync(machine, machine.input, undefined), args)
+      : [decodeInputSync(machine, machine.input, args[0])]
+    const initial = machine.initial(...inputArgs as any)
+    const active = normalizeConfigurationSync(machine, initial as Machine.Snapshot<any>)
+    validateInitialConfiguration(machine, active)
+    const completed = completeConfigurationSync(machine, active, InitialEvent).configuration
+    const configuration = indexedConfigurationFromActive(indexed, completed)
+    const state = snapshotFromIndexed(indexed, configuration)
+    const done = isActiveFinalConfiguration(machine, completed)
+    if (!done) {
+      return {
+        state,
+        configuration,
+        activeConfiguration: completed,
+        initialEntryPaths: getInitialEntryPaths(machine, completed),
+        done: false,
+        output: undefined
+      }
+    }
+    const root = getRootPath(machine, completed)
+    if (!completed.outputs.has(root)) {
+      throw new Error("Machine reached a terminal configuration without a completed root output")
+    }
+    return {
+      state,
+      configuration,
+      activeConfiguration: completed,
+      initialEntryPaths: getInitialEntryPaths(machine, completed),
+      done: true,
+      output: completed.outputs.get(root)
+    }
+  }
+})
+
+export type ExecutionPlanStrategy = "generic" | "indexed-flat" | "indexed-hierarchical" | "auto"
+
+export interface SelectedExecutionPlan {
+  readonly strategy: Exclude<ExecutionPlanStrategy, "auto">
+  readonly plan: CompiledExecutionPlan
+}
+
+const selectExecutionPlan = (
+  machine: Machine.Any,
+  strategy: ExecutionPlanStrategy
+): SelectedExecutionPlan => {
+  if (strategy === "generic") {
+    return { strategy, plan: makeActiveExecutionPlan(machine) }
+  }
+  const indexed = compileIndexedExecutionDescriptor(machine)
+  if (indexed === undefined) {
+    if (strategy === "auto") {
+      return { strategy: "generic", plan: makeActiveExecutionPlan(machine) }
+    }
+    throw new Error(`Machine cannot compile the requested ${strategy} execution plan`)
+  }
+  const selected = indexed.flat ? "indexed-flat" : "indexed-hierarchical"
+  if (strategy !== "auto" && strategy !== selected) {
+    throw new Error(`Machine compiled ${selected}, not the requested ${strategy} execution plan`)
+  }
+  return { strategy: selected, plan: makeIndexedExecutionPlan(machine, indexed) }
+}
+
+/** @internal Test-only uncached strategy selection. */
+export const selectExecutionPlanForTesting = (
+  machine: Machine.Any,
+  strategy: ExecutionPlanStrategy
+): SelectedExecutionPlan => selectExecutionPlan(machine, strategy)
+
 export const compileExecutionPlan = (machine: Machine.Any): CompiledExecutionPlan => {
   const cached = executionPlanCache.get(machine)
   if (cached !== undefined) {
     return cached
   }
   const indexed = compileIndexedExecutionDescriptor(machine)
-  const activePlan = (
-    plan: (configuration: ActiveConfiguration, event: unknown) => MacrostepPlan<ActiveConfiguration, any, any, any, any>
-  ): CompiledExecutionPlan => ({
-    fromConfiguration: (configuration) => configuration,
-    toConfiguration: (state) => state as ActiveConfiguration,
-    snapshot: (state) => snapshotFromConfiguration(machine, state as ActiveConfiguration),
-    plan: (state, event) => plan(state as ActiveConfiguration, event)
-  })
-  const compiled: CompiledExecutionPlan = indexed !== undefined
-    ? {
-      fromConfiguration: (configuration) => indexedConfigurationFromActive(indexed, configuration),
-      toConfiguration: (state) => activeConfigurationFromIndexed(indexed, state as IndexedConfiguration),
-      snapshot: (state) => snapshotFromIndexed(indexed, state as IndexedConfiguration),
-      plan: (state, event) => planIndexedConfiguration(machine, indexed, state as IndexedConfiguration, event),
-      initial: (args) => {
-        const inputArgs = machine.input === undefined
-          ? args
-          : args.length === 0
-          ? (decodeInputSync(machine, machine.input, undefined), args)
-          : [decodeInputSync(machine, machine.input, args[0])]
-        const initial = machine.initial(...inputArgs as any)
-        const active = normalizeConfigurationSync(machine, initial as Machine.Snapshot<any>)
-        validateInitialConfiguration(machine, active)
-        const completed = completeConfigurationSync(machine, active, InitialEvent).configuration
-        const configuration = indexedConfigurationFromActive(indexed, completed)
-        const state = snapshotFromIndexed(indexed, configuration)
-        const done = isActiveFinalConfiguration(machine, completed)
-        if (!done) {
-          return {
-            state,
-            configuration,
-            activeConfiguration: completed,
-            initialEntryPaths: getInitialEntryPaths(machine, completed),
-            done: false,
-            output: undefined
-          }
-        }
-        const root = getRootPath(machine, completed)
-        if (!completed.outputs.has(root)) {
-          throw new Error("Machine reached a terminal configuration without a completed root output")
-        }
-        return {
-          state,
-          configuration,
-          activeConfiguration: completed,
-          initialEntryPaths: getInitialEntryPaths(machine, completed),
-          done: true,
-          output: completed.outputs.get(root)
-        }
-      }
-    }
-    : activePlan((configuration, event) => planConfiguration(machine as any, configuration, event as any))
+  const compiled = indexed === undefined
+    ? makeActiveExecutionPlan(machine)
+    : makeIndexedExecutionPlan(machine, indexed)
   executionPlanCache.set(machine, compiled)
   return compiled
 }

src/internal/machineProcess.ts

@@ -978,6 +978,30 @@ const toResumedProcessLogic = (
 ): internalRuntime.ProcessLogic<any, any, any, any, any, any> =>
   (makeProcessLogic as any)(machine, { _tag: "Resume", snapshot })
 
+/** @internal Test-only runtime strategy selection for a fresh machine. */
+export const startWithRuntimeStrategyForTesting = (
+  machine: Machine.Any,
+  strategy: internalRuntime.ProcessRuntimeStrategy,
+  ...args: ReadonlyArray<unknown>
+): Effect.Effect<internalRuntime.MachineRef<any, any, any, any>, any, any> =>
+  internalRuntime.startProcessWithStrategyForTesting(
+    (toProcessLogic as any)(machine, ...args),
+    strategy,
+    machine.id === undefined ? undefined : { id: machine.id }
+  )
+
+/** @internal Test-only runtime strategy selection for a resumed machine. */
+export const resumeWithRuntimeStrategyForTesting = (
+  machine: Machine.Any,
+  snapshot: Machine.Snapshot<any>,
+  strategy: internalRuntime.ProcessRuntimeStrategy
+): Effect.Effect<internalRuntime.MachineRef<any, any, any, any>, any, any> =>
+  internalRuntime.startProcessWithStrategyForTesting(
+    toResumedProcessLogic(machine, snapshot),
+    strategy,
+    machine.id === undefined ? undefined : { id: machine.id }
+  )
+
 export const start: <
   const States extends Machine.StateSchemas,
   const Events extends ReadonlyArray<Machine.TaggedSchema>,