Reuse configurations while draining events (#40)

Author: SandroMaglione

.changeset/calm-machines-batch.md

@@ -0,0 +1,5 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Reuse the validated statechart configuration while draining queued event batches, avoiding repeated snapshot normalization without retaining the cache while a machine is idle. Canonicalize history snapshot paths in machine document order so batched and public planning produce identical snapshots.

src/internal/machineModel.ts

@@ -215,6 +215,7 @@ const historyFromSnapshotEffect = Effect.fnUntraced(function*(
 })
 
 const historyToSnapshot = (
+  machine: Machine.Any,
   history: ReadonlyMap<string, HistoryRecord>
 ): Readonly<
   Record<string, {
@@ -229,10 +230,11 @@ const historyToSnapshot = (
     readonly values: Readonly<Record<string, unknown>>
   }> = {}
   for (const [path, record] of history) {
+    const active = Array.from(record.active).sort((left, right) => compareDocumentOrder(machine, left, right))
     entries[path] = {
       mode: record.mode,
-      active: Array.from(record.active),
-      values: Object.fromEntries(record.values)
+      active,
+      values: Object.fromEntries(active.map((path) => [path, record.values.get(path)]))
     }
   }
   return entries
@@ -1009,7 +1011,7 @@ export const snapshotFromConfiguration = <const States extends Machine.StateSche
     }).completed = completed
   }
   if (configuration.history.size > 0) {
-    Object.assign(snapshot, { history: historyToSnapshot(configuration.history) })
+    Object.assign(snapshot, { history: historyToSnapshot(machine, configuration.history) })
   }
   return snapshot
 }
@@ -1024,7 +1026,7 @@ export const snapshotFromConfigurationAtPath = <const States extends Machine.Sta
 ): Machine.SnapshotByIdentifier<States, Machine.StateIdentifier<States>> => {
   const snapshot = snapshotFromPath<States>(machine, configuration, path)
   if (configuration.history.size > 0) {
-    Object.assign(snapshot, { history: historyToSnapshot(configuration.history) })
+    Object.assign(snapshot, { history: historyToSnapshot(machine, configuration.history) })
   }
   return snapshot
 }

src/internal/machinePlanner.ts

@@ -2107,7 +2107,7 @@ const settle: <
     : { ...result, done: false, output: undefined }
 })
 
-const macrostep: <
+const macrostepConfiguration: <
   const States extends Machine.StateSchemas,
   const Events extends ReadonlyArray<Machine.TaggedSchema>,
   const Emits extends ReadonlyArray<Machine.TaggedSchema> = readonly [],
@@ -2121,10 +2121,10 @@ const macrostep: <
   Output = never
 >(
   machine: Machine<States, Events, Input, UnhandledStates, E, R, InitialE, InitialR, FinalStates, Output, Emits>,
-  state: Machine.Snapshot<States>,
+  configuration: ActiveConfiguration,
   event: Machine.EventOf<Events>
 ) => Effect.Effect<
-  MacrostepPlan<Machine.Snapshot<States>, Machine.EventOf<Events>, E, R, Output>,
+  MacrostepPlan<ActiveConfiguration, Machine.EventOf<Events>, E, R, Output>,
   E | InfiniteTransitionError | MachineSchemaDecodeError,
   R
 > = Effect.fnUntraced(function*<
@@ -2141,10 +2141,9 @@ const macrostep: <
   Output = never
 >(
   machine: Machine<States, Events, Input, UnhandledStates, E, R, InitialE, InitialR, FinalStates, Output, Emits>,
-  state: Machine.Snapshot<States>,
+  configuration: ActiveConfiguration,
   event: Machine.EventOf<Events>
 ) {
-  const configuration = yield* normalizeConfigurationEffect<States>(machine, state)
   const decodedEvent = yield* decodeEvent<Events>(machine, event)
   if (isActiveFinalConfiguration(machine, configuration)) {
     const completed = yield* completeConfigurationEffect(machine, configuration, decodedEvent)
@@ -2155,7 +2154,7 @@ const macrostep: <
       )
     }
     return {
-      next: snapshotFromConfiguration<States>(machine, completed.configuration),
+      next: completed.configuration,
       actions: [],
       emittedEvents: [],
       microsteps: [],
@@ -2171,7 +2170,7 @@ const macrostep: <
   )
   if (selections.length === 0) {
     return {
-      next: snapshotFromConfiguration<States>(machine, configuration),
+      next: configuration,
       actions: [],
       emittedEvents: [],
       microsteps: [],
@@ -2189,7 +2188,19 @@ const macrostep: <
   const raisedEvents = [...step.raisedEvents]
   const emittedEvents = [...step.emittedEvents]
   const microsteps = [step]
-  const settled = yield* settle(machine, step.next, decodedEvent, actions, raisedEvents, emittedEvents, microsteps)
+  return yield* settle(machine, step.next, decodedEvent, actions, raisedEvents, emittedEvents, microsteps)
+})
+
+const snapshotMacrostep = <
+  const States extends Machine.StateSchemas,
+  Event,
+  E,
+  R,
+  Output
+>(
+  machine: Machine.Any,
+  settled: MacrostepPlan<ActiveConfiguration, Event, E, R, Output>
+): MacrostepPlan<Machine.Snapshot<States>, Event, E, R, Output> => {
   const planned = {
     next: snapshotFromConfiguration<States>(machine, settled.next),
     actions: settled.actions,
@@ -2209,10 +2220,54 @@ const macrostep: <
   return settled.done
     ? { ...planned, done: true, output: settled.output }
     : { ...planned, done: false, output: undefined }
+}
+
+const macrostep: <
+  const States extends Machine.StateSchemas,
+  const Events extends ReadonlyArray<Machine.TaggedSchema>,
+  const Emits extends ReadonlyArray<Machine.TaggedSchema> = readonly [],
+  const Input extends Schema.Top = typeof Schema.Void,
+  UnhandledStates extends Machine.StateIdentifier<States> = Machine.StateIdentifier<States>,
+  E = never,
+  R = never,
+  InitialE = never,
+  InitialR = never,
+  FinalStates extends Machine.StateIdentifier<States> = never,
+  Output = never
+>(
+  machine: Machine<States, Events, Input, UnhandledStates, E, R, InitialE, InitialR, FinalStates, Output, Emits>,
+  state: Machine.Snapshot<States>,
+  event: Machine.EventOf<Events>
+) => Effect.Effect<
+  MacrostepPlan<Machine.Snapshot<States>, Machine.EventOf<Events>, E, R, Output>,
+  E | InfiniteTransitionError | MachineSchemaDecodeError,
+  R
+> = Effect.fnUntraced(function*<
+  const States extends Machine.StateSchemas,
+  const Events extends ReadonlyArray<Machine.TaggedSchema>,
+  const Emits extends ReadonlyArray<Machine.TaggedSchema> = readonly [],
+  const Input extends Schema.Top = typeof Schema.Void,
+  UnhandledStates extends Machine.StateIdentifier<States> = Machine.StateIdentifier<States>,
+  E = never,
+  R = never,
+  InitialE = never,
+  InitialR = never,
+  FinalStates extends Machine.StateIdentifier<States> = never,
+  Output = never
+>(
+  machine: Machine<States, Events, Input, UnhandledStates, E, R, InitialE, InitialR, FinalStates, Output, Emits>,
+  state: Machine.Snapshot<States>,
+  event: Machine.EventOf<Events>
+) {
+  const configuration = yield* normalizeConfigurationEffect<States>(machine, state)
+  const settled = yield* macrostepConfiguration(machine, configuration, event)
+  return snapshotMacrostep<States, Machine.EventOf<Events>, E, R, Output>(machine, settled)
 })
 
 export const plan = macrostep
 
+export const planConfiguration = macrostepConfiguration
+
 const actionUnsafe = Effect.fnUntraced(function*<E, R>(
   effect: Effect.Effect<void, E, R>
 ) {

src/internal/machineProcess.ts

@@ -8,6 +8,7 @@ import * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 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"
@@ -147,14 +148,14 @@ const makeProcessLogic: <
     run: (context) =>
       internalRuntime.provideMachineRuntime(
         Effect.gen(function*() {
-          const { receive, state, setState } = context
+          const { mailbox, receive, state, setState } = context
           let terminal: { readonly output: Output } | undefined
 
-          const initialState = yield* state
-          if (internalPlanner.isFinalState(machine, initialState)) {
+          let current = yield* state
+          if (internalPlanner.isFinalState(machine, current)) {
             return yield* internalPlanner.getFinalOutputEffect<States, Events, Output>(
               machine,
-              initialState,
+              current,
               internalPlanner.InitialEvent
             )
           }
@@ -165,26 +166,46 @@ const makeProcessLogic: <
           )
 
           if (!hasInvokes) {
+            // A queued batch is produced entirely by this worker, so its
+            // configuration is already validated. Drop both caches before
+            // blocking again so idle machines retain only the public snapshot.
+            let configuration: Model.ActiveConfiguration | undefined
+            let pendingEvent: Option.Option<Machine.EventOf<Events>> = Option.none()
+            let pollEvent: Effect.Effect<Option.Option<Machine.EventOf<Events>>> | undefined
             yield* Effect.whileLoop({
               while: () => terminal === undefined,
               body: () =>
                 Effect.gen(function*() {
-                  const event = yield* receive
-                  const current = yield* state
-                  const planned = yield* internalPlanner.plan(machine, current, event)
-                  if (planned.microsteps.length === 0) {
-                    return
-                  }
-
-                  yield* internalPlanner.runActions(planned.actions, liveRuntime)
-                  yield* setState(planned.next)
-                  yield* internalPlanner.runEmittedEvents(
-                    planned.emittedEvents as ReadonlyArray<Machine.EmitOf<Emits>>,
-                    liveRuntime
+                  const event = Option.isSome(pendingEvent) ? pendingEvent.value : yield* receive
+                  pendingEvent = Option.none()
+                  const planned = yield* internalPlanner.planConfiguration(
+                    machine,
+                    configuration ?? (yield* Model.normalizeConfigurationEffect(machine, current)),
+                    event
                   )
+                  configuration = planned.next
+
+                  if (planned.microsteps.length > 0) {
+                    const next = Model.snapshotFromConfiguration<States>(machine, planned.next)
+                    yield* internalPlanner.runActions(planned.actions, liveRuntime)
+                    yield* setState(next)
+                    current = next
+                    yield* internalPlanner.runEmittedEvents(
+                      planned.emittedEvents as ReadonlyArray<Machine.EmitOf<Emits>>,
+                      liveRuntime
+                    )
 
-                  if (planned.done) {
-                    terminal = { output: planned.output }
+                    if (planned.done) {
+                      terminal = { output: planned.output }
+                    }
+                  }
+
+                  if (terminal === undefined) {
+                    pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox))
+                    if (Option.isNone(pendingEvent)) {
+                      configuration = undefined
+                      pollEvent = undefined
+                    }
                   }
                 }),
               step: () => undefined
@@ -349,15 +370,14 @@ const makeProcessLogic: <
             yield* startInvokeWatchers(config, child, key, token, scope)
           })
           const startInvokes: (
-            state: Machine.Snapshot<States>,
+            configuration: Model.ActiveConfiguration,
             paths: ReadonlyArray<string>,
             event: Machine.LifecycleEvent<Events>
           ) => Effect.Effect<void, E | MachineSchemaDecodeError, R> = Effect.fnUntraced(function*(
-            state: Machine.Snapshot<States>,
+            configuration: Model.ActiveConfiguration,
             paths: ReadonlyArray<string>,
             event: Machine.LifecycleEvent<Events>
           ) {
-            const configuration = yield* Model.normalizeConfigurationEffect(machine, state)
             yield* Effect.all(
               internalPlanner.sortEntryPaths(machine, paths)
                 .filter((path) => configuration.active.has(path))
@@ -393,53 +413,76 @@ const makeProcessLogic: <
             )
 
           return yield* Effect.gen(function*() {
+            let configuration: Model.ActiveConfiguration | undefined = yield* Model.normalizeConfigurationEffect(
+              machine,
+              current
+            )
             yield* startInvokes(
-              initialState,
-              Model.getInitialEntryPaths(machine, yield* Model.normalizeConfigurationEffect(machine, initialState)),
+              configuration,
+              Model.getInitialEntryPaths(machine, configuration),
               internalPlanner.InitialEvent
             )
+            // As above, keep the normalized configuration only while this
+            // worker can continue draining an already queued batch.
+            configuration = undefined
+            let pendingEvent: Option.Option<Machine.EventOf<Events>> = Option.none()
+            let pollEvent: Effect.Effect<Option.Option<Machine.EventOf<Events>>> | undefined
 
             yield* Effect.whileLoop({
               while: () => terminal === undefined,
               body: () =>
                 Effect.gen(function*() {
-                  const event = yield* receive
-                  const current = yield* state
-                  const planned = yield* internalPlanner.plan(machine, current, event)
-                  if (planned.microsteps.length === 0) {
-                    return
-                  }
-                  const changed = planned.microsteps.some((step) => step.changed)
-                  const exitPaths = planned.microsteps.flatMap((step) => step.exitPaths)
-                  const entryEvents = new Map<string, Machine.LifecycleEvent<Events>>()
-                  for (const step of planned.microsteps) {
-                    if (step.changed) {
-                      for (const path of step.entryPaths) {
-                        entryEvents.set(path, step.event as Machine.LifecycleEvent<Events>)
+                  const event = Option.isSome(pendingEvent) ? pendingEvent.value : yield* receive
+                  pendingEvent = Option.none()
+                  const planned = yield* internalPlanner.planConfiguration(
+                    machine,
+                    configuration ?? (yield* Model.normalizeConfigurationEffect(machine, current)),
+                    event
+                  )
+                  configuration = planned.next
+                  if (planned.microsteps.length > 0) {
+                    const changed = planned.microsteps.some((step) => step.changed)
+                    const exitPaths = planned.microsteps.flatMap((step) => step.exitPaths)
+                    const entryEvents = new Map<string, Machine.LifecycleEvent<Events>>()
+                    for (const step of planned.microsteps) {
+                      if (step.changed) {
+                        for (const path of step.entryPaths) {
+                          entryEvents.set(path, step.event as Machine.LifecycleEvent<Events>)
+                        }
                       }
                     }
-                  }
-
-                  yield* internalPlanner.runActions(planned.actions, liveRuntime)
-                  if (changed) {
-                    yield* stopInvokes(exitPaths)
-                  }
-                  yield* setState(planned.next)
-                  yield* internalPlanner.runEmittedEvents(
-                    planned.emittedEvents as ReadonlyArray<Machine.EmitOf<Emits>>,
-                    liveRuntime
-                  )
 
-                  if (planned.done) {
-                    terminal = { output: planned.output }
-                    yield* stopAllInvokes(Exit.succeed(planned.output))
-                  } else {
+                    const next = Model.snapshotFromConfiguration<States>(machine, planned.next)
+                    yield* internalPlanner.runActions(planned.actions, liveRuntime)
                     if (changed) {
-                      for (const [path, entryEvent] of entryEvents) {
-                        yield* startInvokes(planned.next, [path], entryEvent)
+                      yield* stopInvokes(exitPaths)
+                    }
+                    yield* setState(next)
+                    current = next
+                    yield* internalPlanner.runEmittedEvents(
+                      planned.emittedEvents as ReadonlyArray<Machine.EmitOf<Emits>>,
+                      liveRuntime
+                    )
+
+                    if (planned.done) {
+                      terminal = { output: planned.output }
+                      yield* stopAllInvokes(Exit.succeed(planned.output))
+                    } else {
+                      if (changed) {
+                        for (const [path, entryEvent] of entryEvents) {
+                          yield* startInvokes(planned.next, [path], entryEvent)
+                        }
                       }
                     }
                   }
+
+                  if (terminal === undefined) {
+                    pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox))
+                    if (Option.isNone(pendingEvent)) {
+                      configuration = undefined
+                      pollEvent = undefined
+                    }
+                  }
                 }),
               step: () => undefined
             })