Use compact compiled mailboxes

Author: SandroMaglione

.changeset/lean-mailboxes-drain.md

@@ -0,0 +1,8 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Use a compact FIFO mailbox for on-demand compiled statecharts while retaining
+Effect Queue for persistent custom process logic. This reduces idle heap for
+machines and invoked families while preserving FIFO delivery, terminal send
+rejection, and wake-up behavior.

src/internal/machineProcess.ts

@@ -153,7 +153,8 @@ const makeProcessLogic: <
     drain: (context: internalRuntime.ProcessContext<Machine.Snapshot<States>, Machine.EventOf<Events>>) =>
       internalRuntime.provideMachineRuntime(
         Effect.gen(function*() {
-          const { mailbox, state, setState } = context
+          const poll = context.poll ?? Queue.poll(context.mailbox!)
+          const { state, setState } = context
           let current = yield* state
           if (internalPlanner.isFinalState(machine, current)) {
             return Option.some(
@@ -380,7 +381,7 @@ const makeProcessLogic: <
           }
 
           while (true) {
-            const pending = yield* Queue.poll(mailbox)
+            const pending = yield* poll
             if (Option.isNone(pending)) {
               return Option.none<Output>()
             }
@@ -447,7 +448,8 @@ const makeProcessLogic: <
     run: (context) =>
       internalRuntime.provideMachineRuntime(
         Effect.gen(function*() {
-          const { mailbox, receive, state, setState } = context
+          const poll = context.poll ?? Queue.poll(context.mailbox!)
+          const { receive, state, setState } = context
           let terminal: { readonly output: Output } | undefined
 
           let current = yield* state
@@ -506,7 +508,7 @@ const makeProcessLogic: <
               }
 
               if (terminal === undefined) {
-                pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox))
+                pendingEvent = yield* (pollEvent ??= poll)
                 if (Option.isNone(pendingEvent)) {
                   configuration = undefined
                   pollEvent = undefined
@@ -785,7 +787,7 @@ const makeProcessLogic: <
               }
 
               if (terminal === undefined) {
-                pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox))
+                pendingEvent = yield* (pollEvent ??= poll)
                 if (Option.isNone(pendingEvent)) {
                   configuration = undefined
                   pollEvent = undefined

src/internal/machineRuntime.ts

@@ -157,14 +157,48 @@ export interface ProcessScope<Event> {
 
 export interface ProcessContext<State, Event> extends ProcessScope<Event> {
   readonly receive: Effect.Effect<Event>
-  readonly mailbox: Queue.Dequeue<Event>
+  /** @internal */
+  readonly mailbox?: Queue.Dequeue<Event>
+  /** @internal */
+  readonly poll?: Effect.Effect<Option.Option<Event>>
   readonly state: Effect.Effect<State>
   readonly setState: (state: State) => Effect.Effect<void>
   readonly updateState: <E, R>(
     f: (state: State) => Effect.Effect<State, E, R>
   ) => Effect.Effect<void, E, R>
 }
 
+interface CompactProcessMailbox<Event> {
+  items: Array<Event> | undefined
+  index: number
+  closed: boolean
+}
+
+const offerCompactMailbox = <Event>(mailbox: CompactProcessMailbox<Event>, event: Event): void => {
+  const items = mailbox.items ?? []
+  mailbox.items = items
+  items.push(event)
+}
+
+const pollCompactMailbox = <Event>(mailbox: CompactProcessMailbox<Event>): Option.Option<Event> => {
+  if (mailbox.items === undefined) {
+    return Option.none()
+  }
+  const event = mailbox.items[mailbox.index]!
+  mailbox.index += 1
+  if (mailbox.index === mailbox.items.length) {
+    mailbox.items = undefined
+    mailbox.index = 0
+  }
+  return Option.some(event)
+}
+
+const closeCompactMailbox = (mailbox: CompactProcessMailbox<unknown>): void => {
+  mailbox.closed = true
+  mailbox.items = undefined
+  mailbox.index = 0
+}
+
 export interface ProcessLogic<
   State,
   Event,
@@ -1145,11 +1179,24 @@ const startCompiledInternal: <
 
   const sessionId = yield* options.runtime.nextSessionId
   const id = options.id ?? sessionId
-  const queue = yield* Queue.unbounded<Event>()
+  const onDemand = logic.drain !== undefined
+  const queue = onDemand ? undefined : yield* Queue.unbounded<Event>()
+  // An on-demand drain never blocks on mailbox input: send schedules its owner
+  // whenever the FIFO becomes non-empty. Keep persistent custom processes on
+  // Queue, but avoid retaining Queue's waiting/backpressure machinery for the
+  // compiled protocol that only needs synchronous offer and poll operations.
+  const compactMailbox: CompactProcessMailbox<Event> | undefined = onDemand
+    ? { items: undefined, index: 0, closed: false }
+    : undefined
+  const poll = compactMailbox === undefined
+    ? Queue.poll(queue!)
+    : Effect.sync(() => pollCompactMailbox(compactMailbox))
+  const shutdownMailbox = compactMailbox === undefined
+    ? Queue.shutdown(queue!)
+    : Effect.sync(() => closeCompactMailbox(compactMailbox))
   const termination = yield* Deferred.make<ProcessTermination>()
   const done = yield* Deferred.make<Output, Error | StoppedError>()
   const awaitCompletion = Deferred.await(done).pipe(Effect.exit, Effect.asVoid)
-  const onDemand = logic.drain !== undefined
   const drainServices = onDemand ? yield* Effect.context<Requirements>() : undefined
   let worker: Fiber.Fiber<any, never> | undefined
   let draining = false
@@ -1158,10 +1205,12 @@ const startCompiledInternal: <
   let terminationRequested = false
   let requestRuntimeTermination = (requested: ProcessTermination): Effect.Effect<boolean> =>
     Deferred.succeed(termination, requested)
-  let sendEvent = (event: Event): Effect.Effect<void, StoppedError> =>
-    Queue.offer(queue, event).pipe(
-      Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError()))
-    )
+  let sendEvent = compactMailbox !== undefined
+    ? (event: Event): Effect.Effect<void, StoppedError> => Effect.sync(() => offerCompactMailbox(compactMailbox, event))
+    : (event: Event): Effect.Effect<void, StoppedError> =>
+      Queue.offer(queue!, event).pipe(
+        Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError()))
+      )
   let settleRequestedTermination = (_requested: ProcessTermination): Effect.Effect<void> => Effect.void
   const interruptWorker: Effect.Effect<void> = Effect.suspend(() =>
     worker === undefined
@@ -1410,7 +1459,7 @@ const startCompiledInternal: <
         Effect.asVoid
       )
     return Effect.uninterruptible(
-      Queue.shutdown(queue).pipe(
+      shutdownMailbox.pipe(
         Effect.andThen(closeChildren(exit)),
         Effect.andThen(setAndPublishSnapshot(snapshot)),
         Effect.andThen(notifyOutcome),
@@ -1536,8 +1585,8 @@ const startCompiledInternal: <
 
   const context: ProcessContext<State, Event> = {
     ...scope,
-    receive: Queue.take(queue),
-    mailbox: queue,
+    receive: queue === undefined ? Effect.never : Queue.take(queue),
+    poll,
     state: getCurrent.pipe(Effect.map((current) => current.snapshot.state)),
     setState: setActiveState,
     updateState: (f) =>
@@ -1683,13 +1732,11 @@ const startCompiledInternal: <
     sendEvent = (event) =>
       Effect.uninterruptible(
         Effect.suspend(() => {
-          if (!Queue.offerUnsafe(queue, event)) {
+          if (compactMailbox!.closed || terminationRequested) {
             return Effect.fail(new StoppedError())
           }
+          offerCompactMailbox(compactMailbox!, event)
           offerRevision += 1
-          if (terminationRequested) {
-            return Effect.fail(new StoppedError())
-          }
           if (draining) {
             return Effect.void
           }

test/MachineProcessLifecycle.test.ts

@@ -1,9 +1,44 @@
 import { assert, describe, it } from "@effect/vitest"
-import { Cause, Deferred, Effect, Exit, Fiber, Option, Queue, Ref, Stream } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Option, Ref, Stream } from "effect"
 import { Machine } from "../src/index.js"
 import * as MachineRuntime from "../src/internal/machineRuntime.js"
 
 describe("machine process lifecycle", () => {
+  it.effect("preserves FIFO order while draining a compact compiled mailbox", () =>
+    Effect.gen(function*() {
+      type Event =
+        | { readonly _tag: "Value"; readonly value: number }
+        | { readonly _tag: "Done" }
+      const observed: Array<number> = []
+      const ref = yield* MachineRuntime.startProcess<undefined, Event, never, never, ReadonlyArray<number>>({
+        [MachineRuntime.childlessProcess]: true,
+        [MachineRuntime.compiledProcess]: true,
+        initial: () => Effect.succeed(undefined),
+        run: () => Effect.never,
+        drain: (context) =>
+          Effect.gen(function*() {
+            while (true) {
+              const event = yield* context.poll!
+              if (Option.isNone(event)) {
+                return Option.none()
+              }
+              if (event.value._tag === "Done") {
+                return Option.some(observed)
+              }
+              observed.push(event.value.value)
+            }
+          })
+      })
+
+      for (let value = 0; value < 1_000; value += 1) {
+        yield* ref.send({ _tag: "Value", value })
+      }
+      yield* ref.send({ _tag: "Done" })
+
+      assert.deepStrictEqual(yield* ref.join, Array.from({ length: 1_000 }, (_, value) => value))
+      assert.instanceOf(yield* Effect.flip(ref.send({ _tag: "Done" })), Machine.StoppedError)
+    }))
+
   it.effect("wakes an on-demand compiled process across consecutive idle periods", () =>
     Effect.gen(function*() {
       type Event = {
@@ -18,7 +53,7 @@ describe("machine process lifecycle", () => {
         drain: (context) =>
           Effect.gen(function*() {
             while (true) {
-              const event = yield* Queue.poll(context.mailbox)
+              const event = yield* context.poll!
               if (Option.isNone(event)) {
                 return Option.none()
               }
@@ -296,7 +331,7 @@ describe("machine process lifecycle", () => {
               run: () => Effect.never,
               drain: (context) =>
                 Effect.gen(function*() {
-                  const event = yield* Queue.poll(context.mailbox)
+                  const event = yield* context.poll!
                   if (Option.isNone(event)) {
                     return Option.none()
                   }