From da648e539925cc97c491264106dc345b5959126e Mon Sep 17 00:00:00 2001 From: SandroMaglione Date: Sat, 8 Aug 2026 14:45:42 +0200 Subject: [PATCH] Compact process supervision --- .changeset/lean-process-supervision.md | 5 + src/internal/machineRuntime.ts | 180 +++++++++++-------------- test/MachineProcessLifecycle.test.ts | 30 +++++ 3 files changed, 115 insertions(+), 100 deletions(-) create mode 100644 .changeset/lean-process-supervision.md diff --git a/.changeset/lean-process-supervision.md b/.changeset/lean-process-supervision.md new file mode 100644 index 0000000..849ad3d --- /dev/null +++ b/.changeset/lean-process-supervision.md @@ -0,0 +1,5 @@ +--- +"@typeonce/effect-machine": patch +--- + +Reduce the memory retained by running machines by consolidating process termination into a single supervisor signal. diff --git a/src/internal/machineRuntime.ts b/src/internal/machineRuntime.ts index cd6046b..283b633 100644 --- a/src/internal/machineRuntime.ts +++ b/src/internal/machineRuntime.ts @@ -10,6 +10,7 @@ import * as Context from "effect/Context" import * as Deferred from "effect/Deferred" 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 PubSub from "effect/PubSub" @@ -304,13 +305,17 @@ const startInternal: < logic: ProcessLogic, options: StartInternalOptions ) { + type ProcessTermination = + | { readonly _tag: "Stopped" } + | { readonly _tag: "Done"; readonly output: Output } + | { readonly _tag: "Failure"; readonly cause: Cause.Cause } + const sessionId = yield* options.runtime.nextSessionId const id = options.id ?? sessionId const queue = yield* Queue.unbounded() - const stopRequested = yield* Deferred.make() - const terminalized = yield* Deferred.make() - const externalFailure = yield* Deferred.make() + const termination = yield* Deferred.make() const done = yield* Deferred.make() + const awaitCompletion = Deferred.await(done).pipe(Effect.exit, Effect.asVoid) const childResourcesState = yield* SynchronizedRef.make({ closed: false, resources: undefined @@ -345,9 +350,7 @@ const startInternal: < const cleanupStartupFailure = (exit: Exit.Exit): Effect.Effect => Exit.isFailure(exit) - ? closeChildren(exit).pipe( - Effect.ensuring(Deferred.succeed(terminalized, void 0)) - ) + ? closeChildren(exit) : Effect.void const cleanup = options.onStop ?? Effect.void @@ -594,7 +597,7 @@ const startInternal: < } let initializing = true - const requestStop = Deferred.succeed(stopRequested, void 0).pipe(Effect.asVoid) + const requestStop = Deferred.succeed(termination, { _tag: "Stopped" }).pipe(Effect.asVoid) const self: ProcessAddress = { id, sessionId, @@ -621,7 +624,11 @@ const startInternal: < sendParent, sendTo, stopChild, - failCause: (cause) => Deferred.failCause(externalFailure, cause as Cause.Cause) + failCause: (cause) => + Deferred.succeed(termination, { + _tag: "Failure", + cause: cause as Cause.Cause + }).pipe(Effect.asVoid) } const initial = yield* logic.initial(scope).pipe( @@ -769,8 +776,7 @@ const startInternal: < Effect.andThen(closeChildren(exit)), Effect.andThen(setAndPublishSnapshot(snapshot)), Effect.andThen(cleanup), - Effect.andThen(completeDone), - Effect.ensuring(Deferred.succeed(terminalized, void 0)) + Effect.andThen(completeDone) ) ) @@ -820,18 +826,8 @@ const startInternal: < return terminalizeWith(snapshot, exit, Deferred.succeed(done, output)) } - const terminalizeStop: Effect.Effect = Effect.uninterruptible( - reserveStoppedSnapshot.pipe( - Effect.flatMap((snapshot) => - snapshot === undefined - ? Deferred.await(terminalized) - : terminalizeReservedStop(snapshot) - ) - ) - ) - const stop: Effect.Effect = Effect.uninterruptible( - requestStop.pipe(Effect.andThen(Deferred.await(terminalized))) + requestStop.pipe(Effect.andThen(awaitCompletion)) ) const context: ProcessContext = { @@ -907,93 +903,77 @@ const startInternal: < yield* options.onReady(ref) } - type ProcessTermination = - | { readonly _tag: "Stopped"; readonly snapshot: RuntimeSnapshot } - | { - readonly _tag: "Done" - readonly snapshot: RuntimeSnapshot - readonly output: Output + const reserveTermination = (termination: ProcessTermination) => { + switch (termination._tag) { + case "Stopped": + return reserveStoppedSnapshot + case "Done": + return reserveSuccessSnapshot(termination.output) + case "Failure": + return reserveFailureSnapshot(termination.cause) } - | { - readonly _tag: "Failure" - readonly snapshot: RuntimeSnapshot - readonly cause: Cause.Cause + } + + const completeTermination = ( + termination: ProcessTermination, + snapshot: RuntimeSnapshot + ) => { + switch (termination._tag) { + case "Stopped": + return terminalizeReservedStop(snapshot) + case "Done": + return terminalizeReservedSuccess(snapshot, termination.output) + case "Failure": + return terminalizeReservedFailure(snapshot, termination.cause) } + } - const arbitration = yield* Deferred.make() - // Only the actual worker and stop waiter are restored to interruptibility. - // Reserving a terminal snapshot, publishing the shared arbitration result, - // and terminalizing stay masked so scope interruption cannot abandon a - // reservation. Both contenders read the same result: a contender that loses - // the reservation cannot finish the race with a different outcome. - const runFiber: Effect.Effect = Effect.uninterruptibleMask((restore) => - Deferred.poll(stopRequested).pipe( - Effect.flatMap((requested) => { - if (Option.isSome(requested)) { - return terminalizeStop - } - const awaitArbitration = Deferred.await(arbitration) - const completeArbitration = (termination: ProcessTermination) => - Deferred.succeed(arbitration, termination).pipe( - Effect.andThen(awaitArbitration) - ) - const stopContender = restore(Deferred.await(stopRequested)).pipe( - Effect.andThen(reserveStoppedSnapshot), - Effect.flatMap((snapshot) => - snapshot === undefined - ? awaitArbitration - : completeArbitration({ _tag: "Stopped", snapshot }) - ) - ) - const workerContender: Effect.Effect = restore( - Effect.raceFirst( - Effect.suspend(() => logic.run(context)), - Deferred.await(externalFailure) - ) - ).pipe( - Effect.exit, - Effect.flatMap((exit) => + const forkRuntime = (effect: Effect.Effect) => + options.fiberScope !== undefined + ? Effect.forkIn(effect, options.fiberScope) + : options.detached === true + ? Effect.forkDetach(effect) + : Effect.forkChild(effect) + + const pendingTermination = yield* Deferred.poll(termination) + const worker = Option.isNone(pendingTermination) + ? yield* Effect.uninterruptibleMask((restore) => + restore(Effect.suspend(() => logic.run(context))).pipe( + Effect.exit, + Effect.flatMap((exit) => + Deferred.succeed( + termination, Exit.isFailure(exit) - ? reserveFailureSnapshot(exit.cause).pipe( - Effect.flatMap((snapshot) => - snapshot === undefined - ? awaitArbitration - : completeArbitration({ _tag: "Failure", snapshot, cause: exit.cause }) - ) - ) - : reserveSuccessSnapshot(exit.value).pipe( - Effect.flatMap((snapshot) => - snapshot === undefined - ? awaitArbitration - : completeArbitration({ _tag: "Done", snapshot, output: exit.value }) - ) - ) + ? { _tag: "Failure", cause: exit.cause } + : { _tag: "Done", output: exit.value } ) ) - return Effect.raceFirst(workerContender, stopContender).pipe( - Effect.flatMap((termination) => { - switch (termination._tag) { - case "Stopped": - return terminalizeReservedStop(termination.snapshot) - case "Done": - return terminalizeReservedSuccess(termination.snapshot, termination.output) - case "Failure": - return terminalizeReservedFailure(termination.snapshot, termination.cause) - } - }) - ) - }) - ) - ) + ) + ).pipe(forkRuntime) + : undefined + + // One Deferred arbitrates all terminal causes. The supervisor reserves the + // terminal snapshot before interrupting the worker, so worker finalizers + // cannot mutate the frozen state. It then waits for those finalizers before + // publishing and completing `join` / `stop`. + const runFiber: Effect.Effect = Effect.uninterruptibleMask((restore) => + Effect.gen(function*() { + const requested = Option.isSome(pendingTermination) + ? yield* pendingTermination.value + : yield* restore(Deferred.await(termination)) - yield* runFiber.pipe( - (effect) => - options.fiberScope !== undefined ? - Effect.forkIn(effect, options.fiberScope) - : options.detached === true ? - Effect.forkDetach(effect) - : Effect.forkChild(effect) + const snapshot = yield* reserveTermination(requested) + if (worker !== undefined) { + yield* Fiber.interrupt(worker) + } + if (snapshot === undefined) { + return yield* awaitCompletion + } + return yield* completeTermination(requested, snapshot) + }) ) + + yield* forkRuntime(runFiber) yield* Effect.yieldNow return ref diff --git a/test/MachineProcessLifecycle.test.ts b/test/MachineProcessLifecycle.test.ts index 3130c43..025fd30 100644 --- a/test/MachineProcessLifecycle.test.ts +++ b/test/MachineProcessLifecycle.test.ts @@ -114,6 +114,36 @@ describe("machine process lifecycle", () => { assert.deepStrictEqual(yield* ref.snapshot, { status: "stopped", state: 1 }) })) + it.effect("interrupts the worker before publishing an externally requested failure", () => + Effect.gen(function*() { + const runtime = yield* Deferred.make>() + const cleanupCount = yield* Ref.make(0) + const logic: MachineRuntime.ProcessLogic = { + initial: (scope) => Deferred.succeed(runtime, scope).pipe(Effect.as(1)), + run: () => + Effect.never.pipe( + Effect.ensuring(Ref.update(cleanupCount, (count) => count + 1)) + ) + } + const ref = yield* MachineRuntime.startProcess( + logic + ) + + yield* (yield* Deferred.await(runtime)).failCause(Cause.fail("external")) + const joined = yield* Effect.exit(ref.join) + + assert.strictEqual(yield* Ref.get(cleanupCount), 1) + assert.deepStrictEqual(yield* ref.snapshot, { + status: "error", + state: 1, + cause: Cause.fail("external") + }) + assert(Exit.isFailure(joined)) + if (Exit.isFailure(joined)) { + assert.strictEqual(joined.cause.reasons.find(Cause.isFailReason)?.error, "external") + } + })) + it.effect("completes a first changes subscription started after terminalization", () => Effect.gen(function*() { const ref = yield* MachineRuntime.startProcess(