Compact compiled terminal coordination

Author: SandroMaglione

.changeset/calm-machines-finish.md

@@ -0,0 +1,5 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Reduce compiled-machine lifecycle and retained-memory overhead by using owner-local terminal arbitration while preserving public completion, cleanup, and first-terminal-wins semantics.

src/internal/machineRuntime.ts

@@ -1194,17 +1194,32 @@ const startCompiledInternal: <
   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 drainServices = onDemand ? yield* Effect.context<Requirements>() : undefined
   let worker: Fiber.Fiber<any, never> | undefined
   let draining = false
   let offerRevision = 0
   let interruptRequested = false
-  let terminationRequested = false
-  let requestRuntimeTermination = (requested: ProcessTermination): Effect.Effect<boolean> =>
-    Deferred.succeed(termination, requested)
+  let requestedTermination: ProcessTermination | undefined
+  let reservedTerminationSnapshot: RuntimeSnapshot<State, Error, Output> | undefined
+  let reserveRequestedTermination:
+    | ((requested: ProcessTermination) => RuntimeSnapshot<State, Error, Output> | undefined)
+    | undefined
+  // A compiled process never suspends waiting for a terminal request: its
+  // active drain owns completion, while stop/failure either interrupts that
+  // owner or terminalizes an idle drain directly. An owner-local cell can
+  // therefore arbitrate the first request atomically without retaining a
+  // second Deferred in every compiled machine.
+  const requestRuntimeTermination = (requested: ProcessTermination): Effect.Effect<boolean> =>
+    Effect.sync(() => {
+      if (requestedTermination !== undefined) {
+        return false
+      }
+      requestedTermination = requested
+      reservedTerminationSnapshot = reserveRequestedTermination?.(requested)
+      return true
+    })
   let sendEvent = compactMailbox !== undefined
     ? (event: Event): Effect.Effect<void, StoppedError> => Effect.sync(() => offerCompactMailbox(compactMailbox, event))
     : (event: Event): Effect.Effect<void, StoppedError> =>
@@ -1408,26 +1423,6 @@ const startCompiledInternal: <
       Effect.asVoid
     )
 
-  const reserveTerminalSnapshot = (
-    f: (
-      snapshot: Extract<RuntimeSnapshot<State, Error, Output>, { readonly status: "active" }>
-    ) => RuntimeSnapshot<State, Error, Output>
-  ): Effect.Effect<RuntimeSnapshot<State, Error, Output> | undefined> =>
-    Effect.sync(() => {
-      const latest = MutableRef.get(current)
-      if (latest.terminalizing || latest.snapshot.status !== "active") {
-        return undefined
-      }
-      const reserved = {
-        revision: latest.revision + 1,
-        snapshot: f(latest.snapshot),
-        terminalizing: true,
-        changes: latest.changes
-      }
-      MutableRef.set(current, { ...latest, terminalizing: true })
-      return reserved.snapshot
-    })
-
   const setAndPublishSnapshot = (
     snapshot: RuntimeSnapshot<State, Error, Output>
   ): Effect.Effect<void> =>
@@ -1469,24 +1464,19 @@ const startCompiledInternal: <
     )
   }
 
-  const reserveStoppedSnapshot = reserveTerminalSnapshot((snapshot) => ({
-    status: "stopped",
-    state: snapshot.state
-  }))
-
-  const reserveFailureSnapshot = (cause: Cause.Cause<Error>) =>
-    reserveTerminalSnapshot((snapshot) => ({
-      status: "error",
-      state: snapshot.state,
-      cause
-    }))
-
-  const reserveSuccessSnapshot = (output: Output) =>
-    reserveTerminalSnapshot((snapshot) => ({
-      status: "done",
-      state: snapshot.state,
-      output
-    }))
+  reserveRequestedTermination = (termination) => {
+    const latest = MutableRef.get(current)
+    if (latest.terminalizing || latest.snapshot.status !== "active") {
+      return undefined
+    }
+    const snapshot: RuntimeSnapshot<State, Error, Output> = termination._tag === "Stopped"
+      ? { status: "stopped", state: latest.snapshot.state }
+      : termination._tag === "Done"
+      ? { status: "done", state: latest.snapshot.state, output: termination.output }
+      : { status: "error", state: latest.snapshot.state, cause: termination.cause }
+    MutableRef.set(current, { ...latest, terminalizing: true })
+    return snapshot
+  }
 
   const terminalizeReservedStop = (
     snapshot: RuntimeSnapshot<State, Error, Output>
@@ -1515,16 +1505,8 @@ const startCompiledInternal: <
     return terminalizeWith(snapshot, exit, Deferred.succeed(done, output))
   }
 
-  const reserveTermination = (termination: ProcessTermination) => {
-    switch (termination._tag) {
-      case "Stopped":
-        return reserveStoppedSnapshot
-      case "Done":
-        return reserveSuccessSnapshot(termination.output)
-      case "Failure":
-        return reserveFailureSnapshot(termination.cause)
-    }
-  }
+  const reserveTermination = (termination: ProcessTermination) =>
+    Effect.sync(() => reserveRequestedTermination!(termination))
 
   const completeTermination = (
     termination: ProcessTermination,
@@ -1540,24 +1522,6 @@ const startCompiledInternal: <
     }
   }
 
-  let reservedTerminationSnapshot: RuntimeSnapshot<State, Error, Output> | undefined
-  requestRuntimeTermination = (requested) =>
-    Deferred.succeed(termination, requested).pipe(
-      Effect.flatMap((accepted) =>
-        accepted
-          ? reserveTermination(requested).pipe(
-            Effect.tap((snapshot) =>
-              Effect.sync(() => {
-                terminationRequested = true
-                reservedTerminationSnapshot = snapshot
-              })
-            ),
-            Effect.as(true)
-          )
-          : Effect.succeed(false)
-      )
-    )
-
   const finishRequestedTermination = (requested: ProcessTermination): Effect.Effect<void> =>
     Effect.gen(function*() {
       const snapshot = reservedTerminationSnapshot ?? (yield* reserveTermination(requested))
@@ -1680,26 +1644,24 @@ const startCompiledInternal: <
       Effect.gen(function*() {
         let observedRevision = offerRevision
         while (true) {
-          const pending = yield* Deferred.poll(termination)
-          if (Option.isSome(pending)) {
-            return yield* finishRequestedTermination(yield* pending.value)
+          if (requestedTermination !== undefined) {
+            return yield* finishRequestedTermination(requestedTermination)
           }
 
           const exit = yield* restore(Effect.suspend(() => logic.drain!(context))).pipe(Effect.exit)
           if (Exit.isFailure(exit)) {
             const completed: ProcessTermination = { _tag: "Failure", cause: exit.cause }
-            yield* Deferred.succeed(termination, completed)
-            return yield* finishRequestedTermination(yield* Deferred.await(termination))
+            yield* requestRuntimeTermination(completed)
+            return yield* finishRequestedTermination(requestedTermination!)
           }
           if (Option.isSome(exit.value)) {
             const completed: ProcessTermination = { _tag: "Done", output: exit.value.value }
-            yield* Deferred.succeed(termination, completed)
-            return yield* finishRequestedTermination(yield* Deferred.await(termination))
+            yield* requestRuntimeTermination(completed)
+            return yield* finishRequestedTermination(requestedTermination!)
           }
 
-          const requested = yield* Deferred.poll(termination)
-          if (Option.isSome(requested)) {
-            return yield* finishRequestedTermination(yield* requested.value)
+          if (requestedTermination !== undefined) {
+            return yield* finishRequestedTermination(requestedTermination)
           }
 
           const continueDraining = yield* Effect.sync(() => {
@@ -1732,7 +1694,7 @@ const startCompiledInternal: <
     sendEvent = (event) =>
       Effect.uninterruptible(
         Effect.suspend(() => {
-          if (compactMailbox!.closed || terminationRequested) {
+          if (compactMailbox!.closed || requestedTermination !== undefined) {
             return Effect.fail(new StoppedError())
           }
           offerCompactMailbox(compactMailbox!, event)
@@ -1763,19 +1725,19 @@ const startCompiledInternal: <
     return ref
   }
 
-  const pendingTermination = yield* Deferred.poll(termination)
+  const pendingTermination = requestedTermination
   const compiledRuntime: Effect.Effect<void, never, Requirements> = Effect.uninterruptibleMask((restore) =>
     Effect.gen(function*() {
       let requested: ProcessTermination
-      if (Option.isSome(pendingTermination)) {
-        requested = yield* pendingTermination.value
+      if (pendingTermination !== undefined) {
+        requested = pendingTermination
       } else {
         const exit = yield* restore(Effect.suspend(() => logic.run(context))).pipe(Effect.exit)
         const completed: ProcessTermination = Exit.isFailure(exit)
           ? { _tag: "Failure", cause: exit.cause }
           : { _tag: "Done", output: exit.value }
-        yield* Deferred.succeed(termination, completed)
-        requested = yield* Deferred.await(termination)
+        yield* requestRuntimeTermination(completed)
+        requested = requestedTermination!
       }
 
       const snapshot = reservedTerminationSnapshot ?? (yield* reserveTermination(requested))

test/MachineProcessLifecycle.test.ts

@@ -169,6 +169,67 @@ describe("machine process lifecycle", () => {
       assert.instanceOf(yield* Effect.flip(ref.join), Machine.StoppedError)
     }))
 
+  it.effect("arbitrates on-demand completion and stop exactly once", () =>
+    Effect.gen(function*() {
+      yield* Effect.forEach(
+        Array.from({ length: 100 }, (_, index) => index),
+        (index) =>
+          Effect.gen(function*() {
+            const started = yield* Deferred.make<void>()
+            const release = yield* Deferred.make<void>()
+            const race = yield* Deferred.make<void>()
+            const cleanupCount = yield* Ref.make(0)
+            const ref = yield* MachineRuntime.startProcess<number, void, never, never, number>({
+              [MachineRuntime.childlessProcess]: true,
+              [MachineRuntime.compiledProcess]: true,
+              initial: () => Effect.succeed(index),
+              run: () => Effect.never,
+              drain: (context) =>
+                Effect.gen(function*() {
+                  const event = yield* context.poll!
+                  if (Option.isNone(event)) {
+                    return Option.none()
+                  }
+                  yield* Deferred.succeed(started, undefined)
+                  return yield* Deferred.await(release).pipe(
+                    Effect.as(Option.some(index)),
+                    Effect.ensuring(Ref.update(cleanupCount, (count) => count + 1))
+                  )
+                })
+            })
+
+            yield* ref.send(undefined)
+            yield* Deferred.await(started)
+            const stopFiber = yield* Deferred.await(race).pipe(
+              Effect.andThen(ref.stop),
+              Effect.forkChild
+            )
+            const completionFiber = yield* Deferred.await(race).pipe(
+              Effect.andThen(Deferred.succeed(release, undefined)),
+              Effect.forkChild
+            )
+
+            yield* Deferred.succeed(race, undefined)
+            yield* Fiber.join(stopFiber)
+            yield* Fiber.join(completionFiber)
+
+            const snapshot = yield* ref.snapshot
+            const joined = yield* Effect.exit(ref.join)
+            assert.strictEqual(yield* Ref.get(cleanupCount), 1)
+            assert(snapshot.status === "done" || snapshot.status === "stopped")
+            if (snapshot.status === "done") {
+              assert.deepStrictEqual(joined, Exit.succeed(index))
+            } else {
+              assert(Exit.isFailure(joined))
+              if (Exit.isFailure(joined)) {
+                assert.instanceOf(joined.cause.reasons.find(Cause.isFailReason)?.error, Machine.StoppedError)
+              }
+            }
+          }),
+        { concurrency: "unbounded" }
+      )
+    }))
+
   it.effect("terminalizes once when concurrent callers stop the same process", () =>
     Effect.gen(function*() {
       const cleanupCount = yield* Ref.make(0)