Durable workflows

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

A workflow is code: a blueprint of nodes (steps, parallel groups, sequences, signal waits, timers) and the queries it answers. With a Pro key the joobq-pro engine conducts every run, as it does a Crystal workflow's: it decides each step (dependencies, retries with backoff, compensation, waits, the workflow timeout) and keeps the run's history in Redis, so a run survives any worker crash.

Define a workflow

A step names a handler and its input. Inputs are references resolved against the run's input and the outputs of the steps before: input(), input("k"), output("step"), output("step", "k"), ref("steps.a.output.k"). A reference to an unknown or a later step is an error when the blueprint is built.

Crystal

workflow "conf-steps-#{LANG}" do
  step "a", StepA, input: {x: input(:x)}
  step "b", StepB, input: {v: ref("steps.a.output.doubled")}
end,

Python

wf("steps").step("a", step_a, input={"x": input("x")}).step("b", step_b, input={"v": output("a", "doubled")}),

TypeScript

wf('steps')
  .step('a', async (i: any, ctx: Ctx) => ({ doubled: i.x * 2, marker: ctx.app.marker, tenant: ctx.context.tenant_id }), { input: { x: input('x') } })
  .step('b', async (i: any) => ({ plus_one: i.v + 1 }), { input: { v: output('a', 'doubled') } }),

Ruby

wf("steps")
  .step("a", ->(i, ctx) { { doubled: i["x"] * 2, marker: ctx.app.marker, tenant: ctx.context["tenant_id"] } }, input: { x: input("x") })
  .step("b", ->(i, _ctx) { { plus_one: i["v"] + 1 } }, input: { v: output("a", "doubled") }),

Go

wf("steps").
	Step("a", func(in map[string]any, ec *EC) (any, error) {
		return map[string]any{"doubled": num(in["x"]) * 2, "marker": ec.App.Marker, "tenant": ec.Meta["tenant_id"]}, nil
	}, joobq.WithInput[*App](map[string]any{"x": joobq.Input("x")})).
	Step("b", func(in map[string]any, ec *EC) (any, error) {
		return map[string]any{"plus_one": num(in["v"]) + 1}, nil
	}, joobq.WithInput[*App](map[string]any{"v": joobq.Output("a", "doubled")})),

Rust

wf("steps")
    .step_with("a", |i: Value, ctx: &Ctx| Ok(json!({"doubled": i["x"].as_i64().unwrap_or(0) * 2, "marker": ctx.app.marker, "tenant": ctx.context.get("tenant_id")})),
        StepOptions { input: json!({"x": input(Some("x"))}), ..Default::default() })
    .step_with("b", |i: Value, _ctx: &Ctx| Ok(json!({"plus_one": i["v"].as_i64().unwrap_or(0) + 1})),
        StepOptions { input: json!({"v": output("a", Some("doubled"))}), ..Default::default() }),

Java

wf("steps")
    .step("a", (i, ctx) -> map("doubled", num(m(i).get("x")) * 2, "marker", ctx.app().marker, "tenant", ctx.context().get("tenant_id")),
        StepOptions.input(map("x", DSL.input("x"))))
    .step("b", (i, ctx) -> map("plus_one", num(m(i).get("v")) + 1), StepOptions.input(map("v", DSL.output("a", "doubled")))),

PHP

wf('steps')
    ->step('a', fn ($i, $ctx) => ['doubled' => $i['x'] * 2, 'marker' => $ctx->app->marker, 'tenant' => $ctx->context['tenant_id'] ?? null], input: ['x' => DSL::input('x')])
    ->step('b', fn ($i, $ctx) => ['plus_one' => $i['v'] + 1], input: ['v' => DSL::output('a', 'doubled')]),

.NET

Wf("steps")
    .Step("a", (i, ctx) => Task.FromResult<object?>(new { doubled = Num(i?["x"]) * 2, marker = ctx.App.Marker, tenant = ctx.Context.GetValueOrDefault("tenant_id")?.GetValue<string>() }),
        input: new { x = Dsl.Input("x") })
    .Step("b", (i, ctx) => Task.FromResult<object?>(new { plus_one = Num(i?["v"]) + 1 }), input: new { v = Dsl.Output("a", "doubled") }),

Tested against a real server in every language shown.

What happens to it

1. When the worker starts, before it registers itself, it sends each workflow's blueprint (POST /sdk/blueprints). The gateway keeps the blueprint's fields, checks them (unique node names, references to steps it has, no reserved names), and versions it by the SHA-256 of its canonical JSON. A known hash changes nothing; a new one is kept beside the old ones (the newest 50) and runs started from then on use it, while runs in flight keep theirs.
2. The gateway compiles the blueprint into an engine definition on the blueprint's task_queue. A start's own task_queue is ignored: the blueprint's applies.
3. The engine sends each step to the workers of that queue as a task of its own: kind: "activity" named <workflow>/<step> (or kind: "compensation", <workflow>/<step>/compensate), with the run's input as payload and the finished steps' outputs as checkpoints. The worker resolves the step's references and runs the handler once, then completes or fails the task like a job.
4. The run's output is the engine's: each step's output by name.

Drawing the diagram…

The engine decides; the worker only runs steps. A worker that dies holding a step loses its lease, and the engine retries the step if it declares retries.

A blueprint whose name a Crystal workflow already uses is refused with 409; a reference to an unknown step, a duplicate name or a query named state or steps is 422.

Start a run

POST /api/v1/sdk/workflows {name, input?, context?, id?, search_attributes?, build?} answers 201 {id, created, status}: a run of the newest registered blueprint (a name no blueprint registered is 422); starting a known id is a no-op (created: false). The SDK methods are start_workflow (Python, Ruby, Rust), startWorkflow (TypeScript, Java, PHP), StartWorkflow (Go) and StartWorkflowAsync (.NET); the Python client also has start(blueprint, input). The console starts runs with POST /api/v1/workflows.

Recovery

A worker that dies holding a step loses its lease, which fails that attempt as retryable (LeaseExpired): the step runs again after a crash only if it declares retries, as in Crystal. One attempt of a step may run up to 24 hours (a dead worker is noticed sooner, by its lease). Every language is tested by killing a worker (SIGKILL) in the middle of a step and restarting it: the finished step is not run again and the interrupted one completes on a later attempt.

Versions that must match

Every process that conducts joobq-pro workflows on the same Redis (Crystal workers included) must run joobq-pro 0.4.1 or later. Queue names starting with wf_activities are the engine's and are refused for jobs, schedules and polls.

Without a Pro key

POST /sdk/blueprints, POST /sdk/workflows, the signal and the client reads answer 402 pro_required; the console's workflow pages and routes are not served. Jobs keep running.