From 0a2ab22f9b111a3f5894752524700c640aa9398a Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Mon, 14 Sep 2026 01:49:02 +0000 Subject: [PATCH 1/4] Fix deferred self-completion in the memory workflow engine --- .../fix-memory-workflow-self-completion.md | 5 + .../src/unstable/workflow/WorkflowEngine.ts | 8 +- .../unstable/workflow/WorkflowEngine.test.ts | 60 ++++++ .../workflow/fixtures/deferred-completion.ts | 171 ++++++++++++++++++ 4 files changed, 243 insertions(+), 1 deletion(-) create mode 100644 .changeset/fix-memory-workflow-self-completion.md create mode 100644 packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts 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..cebbebf66f0 100644 --- a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts +++ b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts @@ -2,6 +2,66 @@ import { assert, describe, it } from "@effect/vitest" import { 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" +import { spawn } from "node:child_process" +import { fileURLToPath } from "node:url" + +describe("deferred completion", () => { + for (const scenario of ["self-success", "self-failure", "plain", "external", "unrelated"] as const) { + it.effect(`settles ${scenario} completion without violating replay ordering`, () => + Effect.gen(function*() { + const result = yield* runDeferredCompletion(scenario) + assert.isFalse(result.timedOut, `deferred completion deadlocked (${scenario}):\n${result.output}`) + assert.strictEqual(result.code, 0, `deferred completion did not settle (${scenario}):\n${result.output}`) + assert.include(result.output, "deferred-completion-passed") + }), 30_000) + } +}) + +// Joining a workflow from its own uninterruptible finalizer can also wedge test +// scope cleanup. Bound the process and wait for it to exit before finishing. +const runDeferredCompletion = (scenario: string) => + Effect.acquireUseRelease( + Effect.sync(() => { + const child = spawn(process.execPath, [ + fileURLToPath(new URL("./fixtures/deferred-completion.ts", import.meta.url)), + scenario + ], { stdio: ["ignore", "pipe", "pipe"] }) + let output = "" + let timedOut = false + let ready = false + const stop = () => { + timedOut = true + child.kill("SIGKILL") + } + let timer = setTimeout(stop, 20_000) + const append = (chunk: Buffer) => { + output = (output + chunk.toString()).slice(-16_000) + if (!ready && output.includes("deferred-completion-ready")) { + ready = true + clearTimeout(timer) + timer = setTimeout(stop, 5_000) + } + } + child.stdout.on("data", append) + child.stderr.on("data", append) + child.on("error", (error) => { + output += String(error) + }) + const closed = new Promise<{ code: number | null; timedOut: boolean; output: string }>((resolve) => { + child.once("close", (code) => { + clearTimeout(timer) + resolve({ code, timedOut, output }) + }) + }) + return { child, closed } + }), + ({ closed }) => Effect.promise(() => closed), + ({ child, closed }) => + Effect.promise(() => { + child.kill("SIGKILL") + return closed + }) + ) describe("WorkflowEngine", () => { const IncrementWorkflow = Workflow.make("WorkflowEngine/IncrementWorkflow", { diff --git a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts new file mode 100644 index 00000000000..255ba94e7fd --- /dev/null +++ b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts @@ -0,0 +1,171 @@ +import { Cause, Effect, Exit, Fiber, Latch, Layer, Option, Schema } from "effect" +import { DurableDeferred, Workflow, WorkflowEngine } from "effect/unstable/workflow" +import * as assert from "node:assert/strict" + +const scenario = process.argv[2] + +const program = Effect.gen(function*() { + const signal = DurableDeferred.make("signal", { success: Schema.String, error: Schema.String }) + const unrelated = DurableDeferred.make("unrelated", { success: Schema.String }) + const read = yield* Latch.make() + const active = yield* Latch.make() + const cleanup = yield* Latch.make() + const releaseCleanup = yield* Latch.make() + const finishActive = yield* Latch.make() + const events: Array = [] + let runs = 0 + let finalizers = 0 + const selfCompletion = scenario === "self-success" || scenario === "self-failure" + const internalCompletion = selfCompletion || scenario === "plain" + const workflow = Workflow.make("DeferredCompletion", { + payload: {}, + success: Schema.String, + error: Schema.String, + idempotencyKey: () => "one" + }) + const layer = workflow.toLayer(() => + Effect.gen(function*() { + const run = ++runs + events.push(`start-${run}`) + if (run === 1) { + yield* Workflow.addFinalizer(() => + Effect.sync(() => { + finalizers++ + events.push("terminal") + }) + ) + } + const engine = yield* WorkflowEngine.WorkflowEngine + const producer = internalCompletion + ? read.await.pipe( + Effect.andThen(Effect.yieldNow), + Effect.tap(() => Effect.sync(() => events.push("completing-signal"))), + Effect.andThen(scenario === "self-failure" ? Effect.fail("boom") : Effect.succeed("ok")), + (effect) => selfCompletion ? DurableDeferred.into(effect, signal) : effect + ) + : active.open.pipe( + Effect.andThen(finishActive.await), + Effect.as("active"), + Effect.onInterrupt(() => + run === 1 + ? Effect.gen(function*() { + events.push("cleanup-start") + yield* cleanup.open + yield* releaseCleanup.await + events.push("cleanup-end") + }) + : Effect.void + ) + ) + return yield* DurableDeferred.raceAll({ + name: "race", + success: Schema.String, + error: Schema.String, + effects: [DurableDeferred.await(signal), producer] + }).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(`body-end-${run}`))) + ) + }) + ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)) + + yield* Effect.gen(function*() { + const executionId = yield* workflow.execute({}, { discard: true }) + yield* read.await + process.stdout.write("deferred-completion-ready\n") + if (!internalCompletion) { + yield* active.await + if (scenario === "external") { + const completion = yield* DurableDeferred.succeed(signal, { + token: DurableDeferred.tokenFromExecutionId(signal, { workflow, executionId }), + value: "ok" + }).pipe(Effect.forkChild) + yield* cleanup.await + for (let i = 0; i < 20; i++) yield* Effect.yieldNow + assert.equal(runs, 1, "replay must wait for the previous body's finalizers") + assert.equal(finalizers, 0, "suspension must preserve the workflow scope") + assert.equal(yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), undefined) + assert.equal(completion.pollUnsafe(), undefined, "external completion must wait for cleanup") + yield* releaseCleanup.open + yield* Fiber.join(completion) + } else { + yield* DurableDeferred.succeed(unrelated, { + token: DurableDeferred.tokenFromExecutionId(unrelated, { workflow, executionId }), + value: "unrelated" + }) + for (let i = 0; i < 20; i++) yield* Effect.yieldNow + assert.deepEqual(events, ["start-1"], "an unrelated completion must not interrupt the active race") + assert.equal(finalizers, 0) + yield* finishActive.open + } + } + + let result = yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)) + for (let i = 0; i < 2_000 && result?._tag !== "Complete"; i++) { + yield* Effect.yieldNow + result = yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)) + } + assert.ok(result?._tag === "Complete", `workflow must complete: ${JSON.stringify(events)}`) + if (scenario === "self-failure") { + assert.ok(Exit.isFailure(result.exit)) + assert.equal(result.exit.cause.reasons.length, 1) + const reason = result.exit.cause.reasons[0] + assert.ok(Cause.isFailReason(reason), "terminal exit must contain only the typed failure") + assert.equal(reason.error, "boom") + } else { + assert.ok(Exit.isSuccess(result.exit)) + assert.equal(result.exit.value, scenario === "unrelated" ? "active" : "ok") + } + for (let run = 2; run <= runs; run++) { + const previousEnd = events.indexOf(`body-end-${run - 1}`) + assert.ok(previousEnd !== -1 && previousEnd < events.indexOf(`start-${run}`), "cleanup must precede replay") + } + assert.equal(finalizers, 1, "terminal finalization must happen exactly once") + if (scenario === "external") { + assert.equal(runs, 2) + assert.deepEqual(events, [ + "start-1", + "cleanup-start", + "cleanup-end", + "body-end-1", + "start-2", + "body-end-2", + "terminal" + ]) + } + const completedRuns = runs + yield* DurableDeferred.succeed(signal, { + token: DurableDeferred.tokenFromExecutionId(signal, { workflow, executionId }), + value: "late" + }) + yield* workflow.resume(executionId) + assert.deepEqual( + yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), + result, + "late completion must preserve the terminal result" + ) + assert.equal(runs, completedRuns) + assert.equal(finalizers, 1) + console.log(JSON.stringify({ scenario, result: "passed", events, runs, finalizers })) + }).pipe(Effect.provide(layer)) +}) + +// An unresolved Effect need not keep Node's event loop alive. Early exit must +// fail too, even when the parent's watchdog never needs to kill the process. +process.exitCode = 1 +Effect.runPromise(program).then( + () => { + process.stdout.write("deferred-completion-passed\n") + process.exitCode = 0 + }, + (error) => { + process.stderr.write(String(error)) + process.exitCode = 1 + } +) From 1e3ef3450a63c95eb7d560d1693806f907cfee15 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Mon, 14 Sep 2026 02:08:30 +0000 Subject: [PATCH 2/4] Require replay in deferred self-completion ordering coverage --- .../unstable/workflow/WorkflowEngine.test.ts | 2 +- .../workflow/fixtures/deferred-completion.ts | 35 ++++++++++++++++--- 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts index cebbebf66f0..8b692e51021 100644 --- a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts +++ b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts @@ -6,7 +6,7 @@ import { spawn } from "node:child_process" import { fileURLToPath } from "node:url" describe("deferred completion", () => { - for (const scenario of ["self-success", "self-failure", "plain", "external", "unrelated"] as const) { + for (const scenario of ["self-success", "self-failure", "self-replay", "plain", "external", "unrelated"] as const) { it.effect(`settles ${scenario} completion without violating replay ordering`, () => Effect.gen(function*() { const result = yield* runDeferredCompletion(scenario) diff --git a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts index 255ba94e7fd..5abc82d6a17 100644 --- a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts +++ b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts @@ -15,7 +15,8 @@ const program = Effect.gen(function*() { const events: Array = [] let runs = 0 let finalizers = 0 - const selfCompletion = scenario === "self-success" || scenario === "self-failure" + const selfReplay = scenario === "self-replay" + const selfCompletion = scenario === "self-success" || scenario === "self-failure" || selfReplay const internalCompletion = selfCompletion || scenario === "plain" const workflow = Workflow.make("DeferredCompletion", { payload: {}, @@ -41,7 +42,24 @@ const program = Effect.gen(function*() { Effect.andThen(Effect.yieldNow), Effect.tap(() => Effect.sync(() => events.push("completing-signal"))), Effect.andThen(scenario === "self-failure" ? Effect.fail("boom") : Effect.succeed("ok")), - (effect) => selfCompletion ? DurableDeferred.into(effect, signal) : effect + (effect) => selfCompletion ? DurableDeferred.into(effect, signal) : effect, + (effect) => + selfReplay + ? effect.pipe( + // Keep the producer alive after recording so only preemption and replay can finish the run. + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + run === 1 + ? Effect.gen(function*() { + events.push("cleanup-start") + yield* cleanup.open + yield* releaseCleanup.await + events.push("cleanup-end") + }) + : Effect.void + ) + ) + : effect ) : active.open.pipe( Effect.andThen(finishActive.await), @@ -79,6 +97,14 @@ const program = Effect.gen(function*() { const executionId = yield* workflow.execute({}, { discard: true }) yield* read.await process.stdout.write("deferred-completion-ready\n") + if (selfReplay) { + yield* cleanup.await + for (let i = 0; i < 20; i++) yield* Effect.yieldNow + assert.equal(runs, 1, "self-completion replay must wait for cleanup") + assert.equal(finalizers, 0, "self-completion must preserve the workflow scope during cleanup") + assert.equal(yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), undefined) + yield* releaseCleanup.open + } if (!internalCompletion) { yield* active.await if (scenario === "external") { @@ -127,10 +153,11 @@ const program = Effect.gen(function*() { assert.ok(previousEnd !== -1 && previousEnd < events.indexOf(`start-${run}`), "cleanup must precede replay") } assert.equal(finalizers, 1, "terminal finalization must happen exactly once") - if (scenario === "external") { - assert.equal(runs, 2) + if (scenario === "external" || selfReplay) { + assert.equal(runs, 2, "completion must replay the workflow") assert.deepEqual(events, [ "start-1", + ...(selfReplay ? ["completing-signal"] : []), "cleanup-start", "cleanup-end", "body-end-1", From a2b8528fe100113da3ed8f28a21220c9a145d7ba Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Mon, 14 Sep 2026 02:37:48 +0000 Subject: [PATCH 3/4] Focus deferred self-completion regression coverage --- .../unstable/workflow/WorkflowEngine.test.ts | 72 ++----- .../workflow/fixtures/deferred-completion.ts | 184 ++++-------------- 2 files changed, 54 insertions(+), 202 deletions(-) diff --git a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts index 8b692e51021..32a2bddb4ba 100644 --- a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts +++ b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts @@ -2,67 +2,25 @@ import { assert, describe, it } from "@effect/vitest" import { 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" -import { spawn } from "node:child_process" +import { execFile } from "node:child_process" import { fileURLToPath } from "node:url" - -describe("deferred completion", () => { - for (const scenario of ["self-success", "self-failure", "self-replay", "plain", "external", "unrelated"] as const) { - it.effect(`settles ${scenario} completion without violating replay ordering`, () => - Effect.gen(function*() { - const result = yield* runDeferredCompletion(scenario) - assert.isFalse(result.timedOut, `deferred completion deadlocked (${scenario}):\n${result.output}`) - assert.strictEqual(result.code, 0, `deferred completion did not settle (${scenario}):\n${result.output}`) - assert.include(result.output, "deferred-completion-passed") - }), 30_000) +import { promisify } from "node:util" + +const exec = promisify(execFile) + +describe("deferred self-completion", () => { + for (const outcome of ["success", "failure"]) { + it.effect(outcome, () => + // A deadlock also wedges Effect scope cleanup, so bound the entire child process. + Effect.promise(() => + exec(process.execPath, [ + fileURLToPath(new URL("./fixtures/deferred-completion.ts", import.meta.url)), + outcome + ], { timeout: 20_000, killSignal: "SIGKILL" }) + ), 30_000) } }) -// Joining a workflow from its own uninterruptible finalizer can also wedge test -// scope cleanup. Bound the process and wait for it to exit before finishing. -const runDeferredCompletion = (scenario: string) => - Effect.acquireUseRelease( - Effect.sync(() => { - const child = spawn(process.execPath, [ - fileURLToPath(new URL("./fixtures/deferred-completion.ts", import.meta.url)), - scenario - ], { stdio: ["ignore", "pipe", "pipe"] }) - let output = "" - let timedOut = false - let ready = false - const stop = () => { - timedOut = true - child.kill("SIGKILL") - } - let timer = setTimeout(stop, 20_000) - const append = (chunk: Buffer) => { - output = (output + chunk.toString()).slice(-16_000) - if (!ready && output.includes("deferred-completion-ready")) { - ready = true - clearTimeout(timer) - timer = setTimeout(stop, 5_000) - } - } - child.stdout.on("data", append) - child.stderr.on("data", append) - child.on("error", (error) => { - output += String(error) - }) - const closed = new Promise<{ code: number | null; timedOut: boolean; output: string }>((resolve) => { - child.once("close", (code) => { - clearTimeout(timer) - resolve({ code, timedOut, output }) - }) - }) - return { child, closed } - }), - ({ closed }) => Effect.promise(() => closed), - ({ child, closed }) => - Effect.promise(() => { - child.kill("SIGKILL") - return closed - }) - ) - describe("WorkflowEngine", () => { const IncrementWorkflow = Workflow.make("WorkflowEngine/IncrementWorkflow", { payload: { value: Schema.Number }, diff --git a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts index 5abc82d6a17..8e725c817c6 100644 --- a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts +++ b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts @@ -1,24 +1,16 @@ -import { Cause, Effect, Exit, Fiber, Latch, Layer, Option, Schema } from "effect" +import { Cause, Effect, Exit, Fiber, Latch, Layer, Schema } from "effect" import { DurableDeferred, Workflow, WorkflowEngine } from "effect/unstable/workflow" import * as assert from "node:assert/strict" -const scenario = process.argv[2] - +const failure = process.argv[2] === "failure" const program = Effect.gen(function*() { const signal = DurableDeferred.make("signal", { success: Schema.String, error: Schema.String }) - const unrelated = DurableDeferred.make("unrelated", { success: Schema.String }) const read = yield* Latch.make() - const active = yield* Latch.make() const cleanup = yield* Latch.make() - const releaseCleanup = yield* Latch.make() - const finishActive = yield* Latch.make() + const release = yield* Latch.make() const events: Array = [] let runs = 0 - let finalizers = 0 - const selfReplay = scenario === "self-replay" - const selfCompletion = scenario === "self-success" || scenario === "self-failure" || selfReplay - const internalCompletion = selfCompletion || scenario === "plain" - const workflow = Workflow.make("DeferredCompletion", { + const workflow = Workflow.make("SelfCompletion", { payload: {}, success: Schema.String, error: Schema.String, @@ -28,58 +20,31 @@ const program = Effect.gen(function*() { Effect.gen(function*() { const run = ++runs events.push(`start-${run}`) - if (run === 1) { - yield* Workflow.addFinalizer(() => - Effect.sync(() => { - finalizers++ - events.push("terminal") - }) - ) - } const engine = yield* WorkflowEngine.WorkflowEngine - const producer = internalCompletion - ? read.await.pipe( - Effect.andThen(Effect.yieldNow), - Effect.tap(() => Effect.sync(() => events.push("completing-signal"))), - Effect.andThen(scenario === "self-failure" ? Effect.fail("boom") : Effect.succeed("ok")), - (effect) => selfCompletion ? DurableDeferred.into(effect, signal) : effect, - (effect) => - selfReplay - ? effect.pipe( - // Keep the producer alive after recording so only preemption and replay can finish the run. - Effect.andThen(Effect.never), - Effect.onInterrupt(() => - run === 1 - ? Effect.gen(function*() { - events.push("cleanup-start") - yield* cleanup.open - yield* releaseCleanup.await - events.push("cleanup-end") - }) - : Effect.void - ) - ) - : effect - ) - : active.open.pipe( - Effect.andThen(finishActive.await), - Effect.as("active"), - Effect.onInterrupt(() => - run === 1 - ? Effect.gen(function*() { - events.push("cleanup-start") - yield* cleanup.open - yield* releaseCleanup.await - events.push("cleanup-end") - }) - : Effect.void - ) - ) return yield* DurableDeferred.raceAll({ name: "race", success: Schema.String, error: Schema.String, - effects: [DurableDeferred.await(signal), producer] + 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, @@ -88,111 +53,40 @@ const program = Effect.gen(function*() { Effect.tap(() => deferred.name === signal.name ? read.open : Effect.void) ) }), - Effect.ensuring(Effect.sync(() => events.push(`body-end-${run}`))) + Effect.ensuring(Effect.sync(() => events.push(`end-${run}`))) ) }) ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)) yield* Effect.gen(function*() { - const executionId = yield* workflow.execute({}, { discard: true }) - yield* read.await - process.stdout.write("deferred-completion-ready\n") - if (selfReplay) { + 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.equal(runs, 1, "self-completion replay must wait for cleanup") - assert.equal(finalizers, 0, "self-completion must preserve the workflow scope during cleanup") - assert.equal(yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), undefined) - yield* releaseCleanup.open - } - if (!internalCompletion) { - yield* active.await - if (scenario === "external") { - const completion = yield* DurableDeferred.succeed(signal, { - token: DurableDeferred.tokenFromExecutionId(signal, { workflow, executionId }), - value: "ok" - }).pipe(Effect.forkChild) - yield* cleanup.await - for (let i = 0; i < 20; i++) yield* Effect.yieldNow - assert.equal(runs, 1, "replay must wait for the previous body's finalizers") - assert.equal(finalizers, 0, "suspension must preserve the workflow scope") - assert.equal(yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), undefined) - assert.equal(completion.pollUnsafe(), undefined, "external completion must wait for cleanup") - yield* releaseCleanup.open - yield* Fiber.join(completion) - } else { - yield* DurableDeferred.succeed(unrelated, { - token: DurableDeferred.tokenFromExecutionId(unrelated, { workflow, executionId }), - value: "unrelated" - }) - for (let i = 0; i < 20; i++) yield* Effect.yieldNow - assert.deepEqual(events, ["start-1"], "an unrelated completion must not interrupt the active race") - assert.equal(finalizers, 0) - yield* finishActive.open - } + assert.deepEqual(events, ["start-1", "cleanup-start"], "replay must wait for cleanup") + yield* release.open } - - let result = yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)) - for (let i = 0; i < 2_000 && result?._tag !== "Complete"; i++) { - yield* Effect.yieldNow - result = yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)) - } - assert.ok(result?._tag === "Complete", `workflow must complete: ${JSON.stringify(events)}`) - if (scenario === "self-failure") { - assert.ok(Exit.isFailure(result.exit)) - assert.equal(result.exit.cause.reasons.length, 1) - const reason = result.exit.cause.reasons[0] - assert.ok(Cause.isFailReason(reason), "terminal exit must contain only the typed failure") + const result = yield* Fiber.join(execution) + if (failure) { + assert.ok(Exit.isFailure(result)) + assert.equal(result.cause.reasons.length, 1) + const reason = result.cause.reasons[0] + assert.ok(Cause.isFailReason(reason)) assert.equal(reason.error, "boom") } else { - assert.ok(Exit.isSuccess(result.exit)) - assert.equal(result.exit.value, scenario === "unrelated" ? "active" : "ok") - } - for (let run = 2; run <= runs; run++) { - const previousEnd = events.indexOf(`body-end-${run - 1}`) - assert.ok(previousEnd !== -1 && previousEnd < events.indexOf(`start-${run}`), "cleanup must precede replay") + assert.deepEqual(result, Exit.succeed("ok")) + assert.deepEqual(events, ["start-1", "cleanup-start", "cleanup-end", "end-1", "start-2", "end-2"]) } - assert.equal(finalizers, 1, "terminal finalization must happen exactly once") - if (scenario === "external" || selfReplay) { - assert.equal(runs, 2, "completion must replay the workflow") - assert.deepEqual(events, [ - "start-1", - ...(selfReplay ? ["completing-signal"] : []), - "cleanup-start", - "cleanup-end", - "body-end-1", - "start-2", - "body-end-2", - "terminal" - ]) - } - const completedRuns = runs - yield* DurableDeferred.succeed(signal, { - token: DurableDeferred.tokenFromExecutionId(signal, { workflow, executionId }), - value: "late" - }) - yield* workflow.resume(executionId) - assert.deepEqual( - yield* workflow.poll(executionId).pipe(Effect.map(Option.getOrUndefined)), - result, - "late completion must preserve the terminal result" - ) - assert.equal(runs, completedRuns) - assert.equal(finalizers, 1) - console.log(JSON.stringify({ scenario, result: "passed", events, runs, finalizers })) }).pipe(Effect.provide(layer)) }) -// An unresolved Effect need not keep Node's event loop alive. Early exit must -// fail too, even when the parent's watchdog never needs to kill the process. +// An unresolved Effect can leave Node's event loop empty. Reject early exit too. process.exitCode = 1 Effect.runPromise(program).then( () => { - process.stdout.write("deferred-completion-passed\n") process.exitCode = 0 }, (error) => { - process.stderr.write(String(error)) - process.exitCode = 1 + console.error(error) } ) From 7e54e809c02e6254fe2e0e284ac69679ed626f62 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Mon, 14 Sep 2026 03:01:39 +0000 Subject: [PATCH 4/4] Run workflow self-completion regressions in process --- .../unstable/workflow/WorkflowEngine.test.ts | 94 ++++++++++++++++--- .../workflow/fixtures/deferred-completion.ts | 92 ------------------ 2 files changed, 79 insertions(+), 107 deletions(-) delete mode 100644 packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts diff --git a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts index 32a2bddb4ba..1da1e95bd39 100644 --- a/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts +++ b/packages/effect/test/unstable/workflow/WorkflowEngine.test.ts @@ -1,23 +1,87 @@ 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" -import { execFile } from "node:child_process" -import { fileURLToPath } from "node:url" -import { promisify } from "node:util" - -const exec = promisify(execFile) describe("deferred self-completion", () => { - for (const outcome of ["success", "failure"]) { - it.effect(outcome, () => - // A deadlock also wedges Effect scope cleanup, so bound the entire child process. - Effect.promise(() => - exec(process.execPath, [ - fileURLToPath(new URL("./fixtures/deferred-completion.ts", import.meta.url)), - outcome - ], { timeout: 20_000, killSignal: "SIGKILL" }) - ), 30_000) + 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) } }) diff --git a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts b/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts deleted file mode 100644 index 8e725c817c6..00000000000 --- a/packages/effect/test/unstable/workflow/fixtures/deferred-completion.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { Cause, Effect, Exit, Fiber, Latch, Layer, Schema } from "effect" -import { DurableDeferred, Workflow, WorkflowEngine } from "effect/unstable/workflow" -import * as assert from "node:assert/strict" - -const failure = process.argv[2] === "failure" -const program = 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.deepEqual(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.equal(result.cause.reasons.length, 1) - const reason = result.cause.reasons[0] - assert.ok(Cause.isFailReason(reason)) - assert.equal(reason.error, "boom") - } else { - assert.deepEqual(result, Exit.succeed("ok")) - assert.deepEqual(events, ["start-1", "cleanup-start", "cleanup-end", "end-1", "start-2", "end-2"]) - } - }).pipe(Effect.provide(layer)) -}) - -// An unresolved Effect can leave Node's event loop empty. Reject early exit too. -process.exitCode = 1 -Effect.runPromise(program).then( - () => { - process.exitCode = 0 - }, - (error) => { - console.error(error) - } -)