Sagas and compensation

ProSagas needs a JoobQ Pro key. Without one every Core feature keeps working: the Pro routes of the SDK gateway answer 402 pro_required and the console's Pro routes are not served. Get a licence · Core and Pro · Pro features index

A step can name a compensation: the handler that undoes it. When a later step fails for good, the compensations of the steps that completed run, newest first; then the run is failed. A compensation's input is the step's output, unless the compensation names an input reference of its own (below, the reservation id).

crystal

workflow "conf-saga-#{LANG}" do
  step "reserve", Reserve, input: {probe: input(:probe)}, compensate: compensation(Release, input: ref("steps.reserve.output.reservation"))
  step "charge", Charge
end,

python

wf("saga").step("reserve", reserve, input={"probe": input("probe")},
                compensate=compensation(release, input=output("reserve", "reservation")))
          .step("charge", charge),

wf('saga')
  .step('reserve', async (i: any) => ({ reservation: `res-${i.probe}` }), {
    input: { probe: input('probe') },
    compensate: compensation(async (i: any) => ({ released: i }), { input: output('reserve', 'reservation') }),
  })
  .step('charge', charge),

ruby

wf("saga")
  .step("reserve", ->(i, _ctx) { { reservation: "res-#{i['probe']}" } }, input: { probe: input("probe") },
                                                                         compensate: compensation(->(i, _ctx) { { released: i } }, input: output("reserve", "reservation")))
  .step("charge", CHARGE),

wf("saga").
	Step("reserve", func(in map[string]any, ec *EC) (any, error) {
		return map[string]any{"reservation": fmt.Sprintf("res-%v", in["probe"])}, nil
	}, joobq.WithInput[*App](map[string]any{"probe": joobq.Input("probe")}),
		joobq.WithCompensate[*App](joobq.Compensation[*App](func(in map[string]any, ec *EC) (any, error) {
			return map[string]any{"released": in["value"]}, nil
		}, joobq.Output("reserve", "reservation")))).
	Step("charge", charge),

rust

wf("saga")
    .step_with("reserve", |i: Value, _ctx: &Ctx| Ok(json!({"reservation": format!("res-{}", i["probe"].as_str().unwrap_or(""))})), StepOptions {
        input: json!({"probe": input(Some("probe"))}),
        compensate: Some(Compensation::new(|i: Value, _ctx: &Ctx| Ok(json!({"released": i})), Some(output("reserve", Some("reservation"))))),
        ..Default::default()
    })
    .step("charge", |_i: Value, _ctx: &Ctx| Err(Error::non_retryable("card declined"))),

java

wf("saga")
    .step("reserve", (i, ctx) -> map("reservation", "res-" + m(i).get("probe")),
        StepOptions.<App>input(map("probe", DSL.input("probe")))
            .compensate(DSL.compensation((i, ctx) -> map("released", i), DSL.output("reserve", "reservation"))))
    .step("charge", Worker::charge),

php

wf('saga')
    ->step('reserve', fn ($i, $ctx) => ['reservation' => 'res-' . $i['probe']], input: ['probe' => DSL::input('probe')],
        compensate: DSL::compensation(fn ($i, $ctx) => ['released' => $i], DSL::output('reserve', 'reservation')))
    ->step('charge', fn ($i, $ctx) => throw new NonRetryableException('card declined')),

csharp

Wf("saga")
    .Step("reserve", (i, ctx) => Task.FromResult<object?>(new { reservation = "res-" + i?["probe"]?.GetValue<string>() }), input: new { probe = Dsl.Input("probe") },
        compensate: Dsl.Compensation<App>((i, ctx) => Task.FromResult<object?>(new { released = i }), Dsl.Output("reserve", "reservation")))
    .Step("charge", (i, ctx) => throw new NonRetryableException("card declined")),

Tested against a real server in every language shown.

Drawing the diagram…

Here charge raises a non-retryable error, so reserve's compensation (release) runs with reserve's reservation and the run fails. The engine sends a compensation to the workers as a task of its own: kind: "compensation", named <workflow>/<step>/compensate.

Cancelling a run from the console (POST /api/v1/workflows/:id/cancel) also undoes the completed steps that can be undone; terminating it (.../terminate) ends it at once. See Operating runs.

Compensating waits, timers and children

A wait (wait_for), a timer or a child node can name a compensation too, written as on a step; its input defaults to the node's own output (the signal's payload, the timer's output, the child run's output). It runs with the others, newest first, when the run fails after the node completed. A skipped node (only_if) is not compensated.