WorkflowEngine.layerMemory: `withCompensation` runs the compensation once per workflow run when a replayed workflow fails

Cerrado
  • ¿Algo no está bien?
#9,202 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Ya se ha fusionado un pull request relacionado.

  • #9239 de @effect-bot — fusionado

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
42/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
typescript
Área
backend

Línea de trabajo

Start with the scope handling in packages/effect/src/workflow/WorkflowEngine.ts around lines 787-791 (layerMemory gives each replay the previous run's scope) and 900 (the interrupt replay), and with Workflow.withCompensation in Workflow.ts around lines 976-980. Run the failing test in packages/effect/test/workflow/WorkflowEngine.test.ts, which the issue includes. Done means the replay case records one refund, matching the cluster engine, and the existing finalizer tests still pass.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

What version of Effect is running?

[email protected] (also main at bc07bd4). Started in 4.0.0-beta.104; 4.0.0-beta.103 runs the compensation once.

What steps can reproduce the bug?

On WorkflowEngine.layerMemory, if a workflow is replayed after a withCompensation step (woken by a DurableDeferred, resumed, or interrupted while suspended) and then fails, the compensation runs once per run of the workflow body. The activity itself runs once.

import { Effect, Layer, Option, Schema } from "effect"
import { Activity, DurableDeferred, Workflow, WorkflowEngine } from "effect/workflow"

const log = []

const Order = Workflow.make("Order", {
  payload: { id: Schema.String },
  error: Schema.String,
  idempotencyKey: ({ id }) => id
})
const Approval = DurableDeferred.make("Approval", { error: Schema.String })

const OrderLayer = Order.toLayer(Effect.fnUntraced(function*() {
  yield* Activity.make({
    name: "charge",
    success: Schema.String,
    execute: Effect.sync(() => {
      log.push("charge")
      return "ch_1"
    })
  }).pipe(Order.withCompensation((id) => Effect.sync(() => log.push(`refund ${id}`))))
  yield* DurableDeferred.await(Approval)
}))

const waitFor = Effect.fnUntraced(function*(executionId, tag) {
  while (true) {
    const result = yield* Order.poll(executionId)
    if (Option.isSome(result) && result.value._tag === tag) return
    yield* Effect.sleep("10 millis")
  }
})

const program = Effect.gen(function*() {
  const executionId = yield* Order.execute({ id: "order-1" }, { discard: true })
  yield* waitFor(executionId, "Suspended")
  // Wakes the workflow; the replay fails it
  yield* DurableDeferred.fail(Approval, {
    token: DurableDeferred.tokenFromExecutionId(Approval, { workflow: Order, executionId }),
    error: "rejected"
  })
  yield* waitFor(executionId, "Complete")
  console.log(log)
})

Effect.runPromise(program.pipe(Effect.provide(OrderLayer.pipe(Layer.provideMerge(WorkflowEngine.layerMemory)))))
What is the expected behavior?
[ 'charge', 'refund ch_1' ]

One compensation for the one completed step ("The compensation finalizer is called if the entire workflow fails"). ClusterWorkflowEngine does this.

What do you see instead?
[ 'charge', 'refund ch_1', 'refund ch_1' ]

The count grows with the number of runs:

  • Calling Order.resume(executionId) before the rejection gives three refunds.
  • Order.interrupt(executionId) in place of the rejection gives two.
  • Failing without suspending gives one.

The same program on ClusterWorkflowEngine.layer (with MessageStorage.layerMemory) gives one in each case.

Additional information
  • withCompensation adds the compensation to instance.scope (via Workflow.addFinalizer) every time the step yields its value, including when a replay reads the cached activity result (Workflow.ts:976-980).
  • Since #7000, layerMemory gives each replay the previous run's scope (WorkflowEngine.ts:787-791). Every run's registration stays in that scope, and closing it on failure runs them all.
  • ClusterWorkflowEngine creates a fresh WorkflowInstance per run (ClusterWorkflowEngine.ts:496), so only the last run's registration fires. The cluster integration suite pins one compensation for this workflow body after it replays on a new owner (cluster-integration/Workflow.test.ts:714).
  • Both engines say a pending interrupt "replays only to rebuild compensations" (WorkflowEngine.ts:900). On layerMemory that replay adds another copy instead.
  • Plain Workflow.addFinalizer differs the same way. On memory every run's finalizer runs ([2, 1], pinned by layerMemory closes finalizers registered before suspension); on cluster only the last run's runs ([2]). Giving the cluster engine the same whole-lifetime scope therefore wouldn't fix this.
Failing test for packages/effect/test/workflow/WorkflowEngine.test.ts (the replay: false case passes) 
for (const replay of [false, true]) {
  it.effect(`layerMemory runs a compensation once when the workflow fails${replay ? " after a replay" : ""}`, () =>
    Effect.gen(function*() {
      const approval = DurableDeferred.make("WorkflowEngine/ReplayCompensation/Approval", {
        error: Schema.String
      })
      const charges: Array<string> = []
      const refunds: Array<string> = []
      let runs = 0
      const Order = Workflow.make("WorkflowEngine/ReplayCompensation", {
        payload: { id: Schema.String },
        error: Schema.String,
        idempotencyKey: ({ id }) => id
      })
      const charge = Activity.make({
        name: "charge",
        success: Schema.String,
        execute: Effect.sync(() => {
          charges.push("charge")
          return "ch_1"
        })
      })
      const layer = Order.toLayer(() =>
        Effect.gen(function*() {
          runs++
          yield* charge.pipe(Order.withCompensation((id) => Effect.sync(() => refunds.push(id))))
          if (!replay) {
            return yield* Effect.fail("rejected")
          }
          yield* DurableDeferred.await(approval)
        })
      ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory))

      yield* Effect.gen(function*() {
        const executionId = yield* Order.execute({ id: "one" }, { discard: true })
        let result = yield* Order.poll(executionId)
        if (replay) {
          while (Option.isNone(result) || result.value._tag !== "Suspended") {
            yield* Effect.yieldNow
            result = yield* Order.poll(executionId)
          }
          const token = DurableDeferred.tokenFromExecutionId(approval, { workflow: Order, executionId })
          yield* DurableDeferred.fail(approval, { token, error: "rejected" })
        }
        while (Option.isNone(result) || result.value._tag !== "Complete") {
          yield* Effect.yieldNow
          result = yield* Order.poll(executionId)
        }

        assert.strictEqual(runs, replay ? 2 : 1)
        assert.deepStrictEqual(charges, ["charge"])
        assert.deepStrictEqual(refunds, ["ch_1"])
      }).pipe(Effect.provide(layer))
    }))
}

On main: AssertionError: expected [ 'ch_1', 'ch_1' ] to deeply equal [ 'ch_1' ].

Related: #9201

Lenguaje dominante
TypeScript
Estrellas
16.7k
Forks
808
Merge medio
11 h 3 min
PR fusionados (30 d)
840

Preparar el entorno

Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de Effect-TS/effect

Todos los issues de Effect-TS/effect

Issues similares

Más issues de TypeScript

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.