Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

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

Chiusa
  • Qualcosa non va?
#9,202 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Una pull request collegata è già stata integrata.

  • #9239 di @effect-bot — integrata

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
42/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
typescript
Ambito
backend

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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

Lingua principale
TypeScript
Stelle
16.7k
Fork
808
Merge medio
11h 24m
PR unite (30g)
848

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di Effect-TS/effect

Tutte le issue di Effect-TS/effect

Issue simili

Altre issue su TypeScript

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.