diff --git a/.changeset/fix-memory-workflow-self-completion.md b/.changeset/fix-memory-workflow-self-completion.md new file mode 100644 index 00000000000..aaf1f81c51c --- /dev/null +++ b/.changeset/fix-memory-workflow-self-completion.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Fix a deadlock in the memory workflow engine when a durable deferred is completed from a finalizer in the workflow awaiting it, including `DurableDeferred.into` inside `DurableDeferred.raceAll`. diff --git a/packages/effect/src/unstable/workflow/WorkflowEngine.ts b/packages/effect/src/unstable/workflow/WorkflowEngine.ts index 46ac7c01b55..e996b5f65ef 100644 --- a/packages/effect/src/unstable/workflow/WorkflowEngine.ts +++ b/packages/effect/src/unstable/workflow/WorkflowEngine.ts @@ -862,10 +862,16 @@ export const layerMemory: Layer.Layer = Layer.effect(WorkflowEng const id = `${options.executionId}/${options.deferredName}` if (deferredResults.has(id)) return Effect.void deferredResults.set(id, options.exit) - return Effect.andThen( + const wake = Effect.andThen( deferredState.deferredDone(options.executionId, options.deferredName, options.exit), resume(options.executionId) ) + return Effect.flatMap(Effect.serviceOption(WorkflowInstance), (instance) => + // A workflow finalizer cannot wait for its own run's cleanup. + // The engine scope owns the wake so cleanup still precedes replay. + Option.isSome(instance) && instance.value.executionId === options.executionId + ? wake.pipe(Effect.forkIn(scope), Effect.asVoid) + : wake) }), scheduleClock: (workflow, options) => engine.deferredDone(options.clock.deferred, { diff --git a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts index 311752b72a4..1da1e95bd39 100644 --- a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts +++ b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts @@ -1,8 +1,90 @@ import { assert, describe, it } from "@effect/vitest" -import { Duration, Effect, Exit, Fiber, Latch, Layer, Option, Ref, Schema, Scope } from "effect" +import { Cause, Duration, Effect, Exit, Fiber, Latch, Layer, Option, Ref, Schema, Scope } from "effect" import { TestClock } from "effect/testing" import { Activity, DurableClock, DurableDeferred, Workflow, WorkflowEngine } from "effect/unstable/workflow" +describe("deferred self-completion", () => { + for (const failure of [false, true]) { + it.live(failure ? "failure" : "success", () => + Effect.gen(function*() { + const signal = DurableDeferred.make("signal", { success: Schema.String, error: Schema.String }) + const read = yield* Latch.make() + const cleanup = yield* Latch.make() + const release = yield* Latch.make() + const events: Array = [] + let runs = 0 + const workflow = Workflow.make("SelfCompletion", { + payload: {}, + success: Schema.String, + error: Schema.String, + idempotencyKey: () => "one" + }) + const layer = workflow.toLayer(() => + Effect.gen(function*() { + const run = ++runs + events.push(`start-${run}`) + const engine = yield* WorkflowEngine.WorkflowEngine + return yield* DurableDeferred.raceAll({ + name: "race", + success: Schema.String, + error: Schema.String, + effects: [ + DurableDeferred.await(signal), + read.await.pipe( + Effect.andThen(Effect.yieldNow), + Effect.andThen(failure ? Effect.fail("boom") : Effect.succeed("ok")), + DurableDeferred.into(signal), + // Successful completion must preempt this producer and replay the run. + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + run === 1 + ? Effect.gen(function*() { + events.push("cleanup-start") + yield* cleanup.open + yield* release.await + events.push("cleanup-end") + }) + : Effect.void + ) + ) + ] + }).pipe( + Effect.provideService(WorkflowEngine.WorkflowEngine, { + ...engine, + deferredResult: (deferred) => + engine.deferredResult(deferred).pipe( + Effect.tap(() => deferred.name === signal.name ? read.open : Effect.void) + ) + }), + Effect.ensuring(Effect.sync(() => events.push(`end-${run}`))) + ) + }) + ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)) + + yield* Effect.gen(function*() { + const execution = yield* workflow.execute({}).pipe(Effect.exit, Effect.forkChild({ startImmediately: true })) + if (!failure) { + yield* cleanup.await + for (let i = 0; i < 20; i++) yield* Effect.yieldNow + assert.deepStrictEqual(events, ["start-1", "cleanup-start"], "replay must wait for cleanup") + yield* release.open + } + const result = yield* Fiber.join(execution) + if (failure) { + assert.ok(Exit.isFailure(result)) + assert.strictEqual(result.cause.reasons.length, 1) + const reason = result.cause.reasons[0] + assert.ok(Cause.isFailReason(reason)) + assert.strictEqual(reason.error, "boom") + } else { + assert.deepStrictEqual(result, Exit.succeed("ok")) + assert.deepStrictEqual(events, ["start-1", "cleanup-start", "cleanup-end", "end-1", "start-2", "end-2"]) + } + }).pipe(Effect.provide(layer)) + }), 5_000) + } +}) + describe("WorkflowEngine", () => { const IncrementWorkflow = Workflow.make("WorkflowEngine/IncrementWorkflow", { payload: { value: Schema.Number },