Conditions, child workflows and continue-as-new

ProWorkflows 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

The engine conducts every run, so these constructs work in every SDK: the worker only runs the steps it is sent.

Conditional steps: only_if

only_if takes a reference. The node (a step, a wait, a timer or a child) runs only when the reference is truthy (not null, false or ""); otherwise it is skipped, and later references to its output read null. The reference may only read steps that run before the node (its dependencies, through order or after:): a parallel sibling or a later step is refused with 422, since it would not have run yet when the engine decides.

Crystal

workflow "conf-cond-#{LANG}" do
  step "check", Check
  step "skipped", Ok, only_if: ref("steps.check.output.go")
  step "ran", Ok, only_if: ref("steps.check.output.yes")
end,

Python

wf("cond").step("check", check).step("skipped", ok, only_if=output("check", "go"))
          .step("ran", ok, only_if=output("check", "yes")),

TypeScript

wf('cond').step('check', check).step('skipped', ok, { onlyIf: output('check', 'go') })
  .step('ran', ok, { onlyIf: output('check', 'yes') }),

Ruby

wf("cond").step("check", CHECK).step("skipped", OK, only_if: output("check", "go"))
          .step("ran", OK, only_if: output("check", "yes")),

Go

wf("cond").
	Step("check", func(map[string]any, *EC) (any, error) { return map[string]any{"go": false, "yes": true}, nil }).
	Step("skipped", ok, joobq.WithOnlyIf[*App](joobq.Output("check", "go"))).
	Step("ran", ok, joobq.WithOnlyIf[*App](joobq.Output("check", "yes"))),

Rust

wf("cond")
    .step("check", check)
    .step_with("skipped", ok, StepOptions { only_if: Some(output("check", Some("go"))), ..Default::default() })
    .step_with("ran", ok, StepOptions { only_if: Some(output("check", Some("yes"))), ..Default::default() }),

Java

wf("cond").step("check", Worker::check)
    .step("skipped", Worker::ok, StepOptions.<App>none().onlyIf(DSL.output("check", "go")))
    .step("ran", Worker::ok, StepOptions.<App>none().onlyIf(DSL.output("check", "yes"))),

PHP

wf('cond')->step('check', 'check')->step('skipped', 'ok', onlyIf: DSL::output('check', 'go'))
    ->step('ran', 'ok', onlyIf: DSL::output('check', 'yes')),

.NET

Wf("cond").Step("check", Check).Step("skipped", Ok, onlyIf: Dsl.Output("check", "go")).Step("ran", Ok, onlyIf: Dsl.Output("check", "yes")),

Tested against a real server in every language shown.

check returns {"go": false, "yes": true}: skipped is skipped and ran runs.

Child workflows: child

A child node starts another registered workflow (an SDK or a Crystal one) and waits for it. Its output is the child run's output; if the child fails, the node fails. Its input is resolved by the engine, not by your code, so it holds only references: a ${...} reference, or objects and lists of them. A literal value is refused (compute a constant in a step and refer to its output). Without input the child gets the parent run's input. The child's steps see the parent's ctx.context.

Crystal

workflow "conf-parent-#{LANG}" do
  step "pre", Pre
  child "kid", "conf-steps-#{LANG}", input: {x: ref("steps.pre.output.x")}
end,

Python

wf("parent").step("pre", pre).child("kid", f"conf-steps-{LANG}", input={"x": output("pre", "x")}),

TypeScript

wf('parent').step('pre', pre).child('kid', `conf-steps-${LANG}`, { input: { x: output('pre', 'x') } }),

Ruby

wf("parent").step("pre", PRE).child("kid", "conf-steps-#{LANG}", input: { x: output("pre", "x") }),

Go

wf("parent").
	Step("pre", func(map[string]any, *EC) (any, error) { return map[string]any{"x": 5}, nil }).
	Child("kid", fmt.Sprintf("conf-steps-%s", lang),
		joobq.WithInput[*App](map[string]any{"x": joobq.Output("pre", "x")})),

Rust

wf("parent")
    .step("pre", pre)
    .child_with("kid", format!("conf-steps-{}", lang), ChildOptions { input: Some(json!({"x": output("pre", Some("x"))})), ..Default::default() }),

Java

wf("parent").step("pre", Worker::pre)
    .child("kid", "conf-steps-" + LANG, ChildOptions.input(map("x", DSL.output("pre", "x")))),

PHP

wf('parent')->step('pre', fn () => ['x' => 5])->child('kid', 'conf-steps-' . LANG, input: ['x' => DSL::output('pre', 'x')]),

.NET

Wf("parent").Step("pre", Pre).Child("kid", $"conf-steps-{Lang}", input: new { x = Dsl.Output("pre", "x") }),

Tested against a real server in every language shown.

The child here is the conf-steps workflow with x = 5; the kid node's output is the child run's output (each of its steps' outputs by name).

Not retried, no own deadline

The engine does not retry or time out a child node, so retry and timeout_ms on a child are refused (422). Give the child workflow's own steps a retry policy, and the child workflow a timeout_ms.

The child workflow must be registered (its worker started) before a run reaches the child node: an unknown child workflow makes the engine quarantine the parent run.

Continue-as-new

continue_as_new closes a run as continued_as_new when it ends (and only_if, if given, is truthy) and starts a fresh run of the same workflow with input (the same input by default). It keeps the history of a workflow that loops for a long time small. max_runs limits the length of the chain; max_events continues a run once its history reaches that many events (then input may only use ${input...}, since later steps may not have run). The next runs keep the first run's ctx.context. A run's continued_to names its successor.

Crystal

workflow "conf-loop-#{LANG}" do
  step "count", Count, input: {n: input(:n)}
  continue_as_new input: {n: ref("steps.count.output.n")}, only_if: ref("steps.count.output.more")
end,

Python

wf("loop").step("count", count, input={"n": input("n")})
          .continue_as_new(input={"n": output("count", "n")}, only_if=output("count", "more")),

TypeScript

wf('loop').step('count', count, { input: { n: input('n') } })
  .continueAsNew({ input: { n: output('count', 'n') }, onlyIf: output('count', 'more') }),

Ruby

wf("loop").step("count", COUNT, input: { n: input("n") })
          .continue_as_new(input: { n: output("count", "n") }, only_if: output("count", "more")),

Go

wf("loop").
	Step("count", func(in map[string]any, ec *EC) (any, error) {
		n := int(num(in["n"])) + 1
		return map[string]any{"n": n, "more": n < 3}, nil
	}, joobq.WithInput[*App](map[string]any{"n": joobq.Input("n")})).
	ContinueAsNew(joobq.ContinueAsNewOptions{
		Input:  map[string]any{"n": joobq.Output("count", "n")},
		OnlyIf: joobq.Output("count", "more"),
	}),

Rust

wf("loop")
    .step_with("count", count, StepOptions { input: json!({"n": input(Some("n"))}), ..Default::default() })
    .continue_as_new(ContinueAsNew {
        input: Some(json!({"n": output("count", Some("n"))})),
        only_if: Some(output("count", Some("more"))),
        ..Default::default()
    }),

Java

wf("loop").step("count", Worker::count, StepOptions.input(map("n", DSL.input("n"))))
    .continueAsNew(ContinueAsNewOptions.input(map("n", DSL.output("count", "n"))).onlyIf(DSL.output("count", "more"))),

PHP

wf('loop')->step('count', 'count_up', input: ['n' => DSL::input('n')])
    ->continueAsNew(input: ['n' => DSL::output('count', 'n')], onlyIf: DSL::output('count', 'more')),

.NET

Wf("loop").Step("count", Count, input: new { n = Dsl.Input("n") })
    .ContinueAsNew(input: new { n = Dsl.Output("count", "n") }, onlyIf: Dsl.Output("count", "more")),

Tested against a real server in every language shown.

count adds one to n while n < 3: three runs, the last one completes with n = 3.

Drawing the diagram…

The chain the example's loop workflow makes. Each finished run's continued_to names the next one; workflow_result answers for the run at the end of the chain (Queries and client reads).