Scopes, updates and change gates

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.

Cancellation scopes: scope

A scope is a group of steps that can be cancelled as a unit while the rest of the run goes on. Its children run in sequence (they may be parallel or sequence groups); scopes do not nest. A scope is cancelled from the console or the API (POST /api/v1/workflows/:id/scopes/:scope/cancel, operator; 404 for an unknown scope, 409 when there is nothing left to cancel), or by its own timeout_ms (at most a year), counted from the moment its first step starts. Then, in order:

1. its steps that have not started, its timers, its signal waits and its child workflows are cancelled at once (a child run is asked to cancel): they read cancelled, with a null output;
2. its steps that are already running are waited for, and keep their own outcome. In them ctx.heartbeat() answers true, so a long step can stop early and return. That needs the step's heartbeat_ms: without it ctx.heartbeat() tells the server nothing, the step runs to its own end (up to its timeout_ms, a day by default) and the scope is cancelled only then. A failure of a step of a cancelled scope no longer fails the workflow;
3. its completed steps that have a compensate are compensated, newest first;
4. the steps after the scope go on: a cancelled step counts as finished.

Crystal

workflow "conf-scoped-#{LANG}" do
  scope "optional" do
    step "long", LongStep, heartbeat: 5.seconds
  end
  step "after", Ok
end,
workflow "conf-shield-#{LANG}" do
  step "charge", Charge
  scope "cleanup", non_cancellable: true do
    step "tidy", Ok
  end
end,

Python

wf("scoped").scope("optional", lambda s: s.step("long", long_step, heartbeat_ms=5000))
            .step("after", ok),
wf("shield").step("charge", charge)
            .scope("cleanup", lambda s: s.step("tidy", ok), non_cancellable=True),

TypeScript

wf('scoped').scope('optional', (s) => { s.step('long', longStep, { heartbeatMs: 5000 }) })
  .step('after', ok),
wf('shield').step('charge', charge)
  .scope('cleanup', (s) => { s.step('tidy', ok) }, { nonCancellable: true }),

Ruby

wf("scoped").scope("optional") { step("long", LONG_STEP, heartbeat_ms: 5000) }
            .step("after", OK),
wf("shield").step("charge", CHARGE)
            .scope("cleanup", non_cancellable: true) { step("tidy", OK) },

Go

wf("scoped").
	Scope("optional", func(s *joobq.WorkflowBlueprint[*App]) { s.Step("long", longStep, joobq.WithHeartbeat[*App](5000)) }).
	Step("after", ok),
wf("shield").
	Step("charge", charge).
	Scope("cleanup", func(s *joobq.WorkflowBlueprint[*App]) { s.Step("tidy", ok) }, joobq.ScopeOptions{NonCancellable: true}),

Rust

wf("scoped")
    .scope("optional", |s| s.step_with("long", long_step, StepOptions { heartbeat_ms: Some(5000), ..Default::default() }))
    .step("after", ok),
wf("shield")
    .step("charge", charge)
    .scope_with("cleanup", ScopeOptions { non_cancellable: true, ..Default::default() }, |s| s.step("tidy", ok)),

Java

wf("scoped").scope("optional", s -> s.step("long", Worker::longStep, StepOptions.<App>none().heartbeatMs(5000)))
    .step("after", Worker::ok),
wf("shield").step("charge", Worker::charge)
    .scope("cleanup", s -> s.step("tidy", Worker::ok), ScopeOptions.none().nonCancellable()),

PHP

wf('scoped')->scope('optional', fn ($s) => $s->step('long', 'long_step', heartbeatMs: 5000))
    ->step('after', 'ok'),
wf('shield')->step('charge', 'charge')
    ->scope('cleanup', fn ($s) => $s->step('tidy', 'ok'), nonCancellable: true),

.NET

Wf("scoped").Scope("optional", s => s.Step("long", LongStep, heartbeatMs: 5000)).Step("after", Ok),
Wf("shield").Step("charge", Charge).Scope("cleanup", s => s.Step("tidy", Ok), nonCancellable: true),

Tested against a real server in every language shown.

Crystal

# Runs until its scope is cancelled: ctx.cancel_requested? turns true once a cancel was asked for.
class LongStep
  def perform(input : JSON::Any, ctx : JoobQPro::Workflows::ActivityContext) : JSON::Any
    150.times do
      ctx.heartbeat
      return Conf.any({stopped: true}) if ctx.cancel_requested?
      sleep 200.milliseconds
    end
    Conf.any({stopped: false})
  end
end

Python

def long_step(inp, ctx):
    # Runs until its scope is cancelled: ctx.heartbeat() answers True once a cancel was asked for.
    for _ in range(150):
        if ctx.heartbeat():
            return {"stopped": True}
        time.sleep(0.2)
    return {"stopped": False}

TypeScript

// Runs until its scope is cancelled: ctx.heartbeat() answers true once a cancel was asked for.
const longStep = async (_i: any, ctx: Ctx) => {
  for (let n = 0; n < 150; n++) {
    if (await ctx.heartbeat()) return { stopped: true }
    await sleep(200)
  }
  return { stopped: false }
}

Ruby

# Runs until its scope is cancelled: ctx.heartbeat answers true once a cancel was asked for.
LONG_STEP = lambda { |_i, ctx|
  150.times do
    return { stopped: true } if ctx.heartbeat

    sleep 0.2
  end
  { stopped: false }
}

Go

// longStep runs until its scope is cancelled: ec.Heartbeat() answers true once a cancel was asked for.
longStep := func(_ map[string]any, ec *EC) (any, error) {
	for range 150 {
		if cancelled, _ := ec.Heartbeat(); cancelled {
			return map[string]any{"stopped": true}, nil
		}
		time.Sleep(200 * time.Millisecond)
	}
	return map[string]any{"stopped": false}, nil
}

Rust

fn long_step(_i: Value, ctx: &Ctx) -> joobq::Result<Value> {
    // Runs until its scope is cancelled: ctx.heartbeat() answers true once a cancel was asked for.
    for _ in 0..150 {
        if ctx.heartbeat()? {
            return Ok(json!({"stopped": true}));
        }
        std::thread::sleep(std::time::Duration::from_millis(200));
    }
    Ok(json!({"stopped": false}))
}

Java

/** Runs until its scope is cancelled: ctx.heartbeat() answers true once a cancel was asked for. */
static Object longStep(Object i, JoobQClient.ExecutionContext<App> ctx) throws Exception {
    for (int n = 0; n < 150; n++) {
        if (ctx.heartbeat()) return map("stopped", true);
        Thread.sleep(200);
    }
    return map("stopped", false);
}

PHP

/** Runs until its scope is cancelled: `$ctx->heartbeat()` answers true once a cancel was asked for. */
function long_step(mixed $i, ExecutionContext $ctx): array
{
    for ($n = 0; $n < 150; $n++) {
        if ($ctx->heartbeat()) {
            return ['stopped' => true];
        }
        sleep_ms(200);
    }
    return ['stopped' => false];
}

.NET

static async Task<object?> LongStep(JsonNode? i, ExecutionContext<App> ctx)
{
    // Runs until its scope is cancelled: ctx.HeartbeatAsync() answers true once a cancel was asked for.
    for (var n = 0; n < 150; n++)
    {
        if (await ctx.HeartbeatAsync()) return new { stopped = true };
        await Task.Delay(200);
    }
    return new { stopped = false };
}

Tested against a real server in every language shown.

Cancelling optional while long runs makes ctx.heartbeat() answer true: long returns {"stopped": true} and ends completed, then after runs.

Cleanup that must run: non_cancellable

A non_cancellable scope is shielded: when the workflow is cancelled, times out or fails, its steps that have not run are still scheduled and finish before the completed steps are compensated. A failed shielded step does not change how the run ends. Its dependencies on steps outside it are abandoned while the workflow unwinds (they do not hold it back), so a cleanup step that reads such a step's output may read null; it also waits until the run's other running steps have stopped. A shielded scope cannot have a timeout_ms (bound its steps' own timeout_ms instead), and cancelling it is refused (422 scope_not_cancellable). In shield above, charge fails, and tidy in the cleanup scope still runs.

after, only_if, compensate and gate go on the scope's steps, not on the scope itself (422).

Update handlers: update

An update is a request sent to a running workflow that the workflow answers: a signal with a reply. Declare a handler on the workflow; send updates with update_workflow from the SDK client (POST /api/v1/sdk/workflows/:id/updates) or from the console (POST /api/v1/workflows/:id/updates/:name).

• validate (a reference) must be truthy, or the update is rejected with reason: "validation_failed" and the handler's message. Truthy is the engine's rule: null, false and "" are falsy, anything else (also 0, [] and {}) is truthy. Other rejections: unknown_update (the workflow declares no such update; this is a rejected outcome, not a 404), workflow_closed, workflow_unwinding (the run is being cancelled or compensated).
• With a handler, the engine runs it as a step of the run (the activity <workflow>/@update/<name>, on the blueprint's task queue). Its payload is input, by default the update's input; its return value is the update's result. It is not retried, and it may take timeout_ms (30 s by default).
• Without a handler, result (a template of references, by default the update's input) is the result, resolved by the engine. result with a handler, input or timeout_ms without one, and message without validate are refused (422): they would do nothing.
• validate, input and result may read the run (${input...}, ${steps.<step>.output...}, another update's outcome with ${updates.<name>...}) and the update itself: ${update.input...}, ${update.name}, ${update.id}. Each SDK has a helper for the update's input: update_input() in Python, Ruby and Rust, updateInput() in TypeScript, UpdateInput() in Go, DSL.updateInput() in Java, DSL::updateInput() in PHP, Dsl.UpdateInput() in .NET; Crystal writes ref("update.input"). They carry references only; message is text stored in the blueprint, so never put runtime data in it.

Crystal

workflow "conf-upd-#{LANG}" do
  wait_for "hold", signal: "go", timeout: 60.seconds
  update "rename", Rename, validate: ref("update.input.name"), message: "name is required"
  update "echo", result: {got: ref("update.input")}
end,

Python

wf("upd").wait_for("hold", signal="go", timeout_ms=60000)
         .update("rename", rename, validate=update_input("name"), message="name is required")
         .update("echo", result={"got": update_input()}),

TypeScript

wf('upd').waitFor('hold', { signal: 'go', timeoutMs: 60000 })
  .update('rename', rename, { validate: updateInput('name'), message: 'name is required' })
  .update('echo', null, { result: { got: updateInput() } }),

Ruby

wf("upd").wait_for("hold", signal: "go", timeout_ms: 60_000)
         .update("rename", RENAME, validate: update_input("name"), message: "name is required")
         .update("echo", result: { got: update_input }),

Go

wf("upd").
	WaitFor("hold", "go", 60000).
	Update("rename", joobq.UpdateOptions[*App]{Handler: rename, Validate: joobq.UpdateInput("name"), Message: "name is required"}).
	Update("echo", joobq.UpdateOptions[*App]{Result: map[string]any{"got": joobq.UpdateInput()}}),

Rust

wf("upd")
    .wait_for("hold", "go", 60_000)
    .update("rename", UpdateOptions { validate: Some(update_input(Some("name"))), message: Some("name is required".into()), ..UpdateOptions::handler(rename) })
    .update("echo", UpdateOptions { result: Some(json!({"got": update_input(None)})), ..Default::default() }),

Java

wf("upd").waitFor("hold", "go", 60_000)
    .update("rename", UpdateOptions.<App>handler(Worker::rename).validate(DSL.updateInput("name")).message("name is required"))
    .update("echo", UpdateOptions.<App>none().result(map("got", DSL.updateInput()))),

PHP

wf('upd')->waitFor('hold', 'go', 60000)
    ->update('rename', 'rename_update', validate: DSL::updateInput('name'), message: 'name is required')
    ->update('echo', result: ['got' => DSL::updateInput()]),

.NET

Wf("upd").WaitFor("hold", "go", 60000)
    .Update("rename", Rename, validate: Dsl.UpdateInput("name"), message: "name is required")
    .Update("echo", result: new { got = Dsl.UpdateInput() }),

Tested against a real server in every language shown.

Crystal

class Rename
  def perform(input : JSON::Any, ctx : JoobQPro::Workflows::ActivityContext) : JSON::Any
    Conf.any({renamed: input["name"]})
  end
end

Python

def rename(inp, ctx):
    return {"renamed": inp["name"]}

TypeScript

const rename = async (i: any) => ({ renamed: i.name })

Ruby

RENAME = ->(i, _ctx) { { renamed: i["name"] } }

Go

rename := func(in map[string]any, _ *EC) (any, error) { return map[string]any{"renamed": in["name"]}, nil }

Rust

fn rename(i: Value, _c: &Ctx) -> joobq::Result<Value> {
    Ok(json!({"renamed": i["name"]}))
}

Java

static Object rename(Object i, JoobQClient.ExecutionContext<App> ctx) {
    return map("renamed", m(i).get("name"));
}

PHP

/** The `upd` workflow's `rename` update handler: its payload is the update's input. */
function rename_update(mixed $i, ExecutionContext $ctx): array
{
    return ['renamed' => $i['name']];
}

.NET

static Task<object?> Rename(JsonNode? i, ExecutionContext<App> c) => Task.FromResult<object?>(new { renamed = i?["name"]?.DeepClone() });

Tested against a real server in every language shown.

Sending an update from the SDK client:

Python

@job("conf.wfupdate", queue=QUEUE)
def wfupdate_job(payload, ctx):
    outcome = ctx.update_workflow(payload["wf_id"], "echo", {"v": payload["probe"]}, wait_ms=10000)
    report(ctx, "wfupdate", payload, status=outcome.get("status"), got=(outcome.get("result") or {}).get("got"))

TypeScript

defineJob<App>({ name: 'conf.wfupdate', queue: QUEUE, perform: async (p, ctx) => {
  const outcome = await ctx.updateWorkflow(p.wf_id, 'echo', { v: p.probe }, { waitMs: 10000 })
  return report(ctx, 'wfupdate', p, { status: outcome.status, got: outcome.result?.got })
} }),

Ruby

class WfUpdateJob
  include JoobQ::Job
  job_name "conf.wfupdate"
  queue QUEUE
  def self.perform(p, ctx)
    outcome = ctx.update_workflow(p["wf_id"], "echo", input: { v: p["probe"] }, wait_ms: 10_000)
    report(ctx, "wfupdate", p, status: outcome["status"], got: (outcome["result"] || {})["got"])
  end
end

Go

joobq.DefineJob[*App]("conf.wfupdate", opts, func(p map[string]any, ec *EC) (any, error) {
	wf, _ := p["wf_id"].(string)
	outcome, err := ec.UpdateWorkflow(wf, "echo", map[string]any{"v": p["probe"]}, "", 10000)
	if err != nil {
		return nil, err
	}
	result, _ := outcome["result"].(map[string]any)
	return report(ec, "wfupdate", p, map[string]any{"status": outcome["status"], "got": result["got"]})
}),

Rust

JobDef::new("conf.wfupdate", queue.clone(), |p: Value, ctx: &Ctx| {
    let wf = p["wf_id"].as_str().unwrap_or("").to_string();
    let outcome = ctx.update_workflow(&wf, "echo", json!({"v": p["probe"]}), None, Some(10_000))?;
    report(ctx, "wfupdate", &p, json!({"status": outcome["status"], "got": outcome["result"]["got"]}))
}),

Java

JobDefinition.<App>of("conf.wfupdate", QUEUE, 3, (p, ctx) -> {
    Map<String, Object> outcome = ctx.updateWorkflow((String) m(p).get("wf_id"), "echo", map("v", m(p).get("probe")), null, 10_000L);
    Object result = outcome.get("result");
    return report(ctx, "wfupdate", p, map("status", outcome.get("status"),
        "got", result instanceof Map<?, ?> r ? r.get("got") : null));
}),

PHP

final class WfUpdateJob extends ConfJob
{
    public static function jobName(): string { return 'conf.wfupdate'; }
    public static function perform(array $p, ExecutionContext $ctx): mixed
    {
        $outcome = $ctx->updateWorkflow($p['wf_id'], 'echo', ['v' => $p['probe']], waitMs: 10000);
        return report($ctx, 'wfupdate', $p, ['status' => $outcome['status'] ?? null, 'got' => $outcome['result']['got'] ?? null]);
    }
}

.NET

JobDefinition<App>.Define("conf.wfupdate", Queue, 3, async (p, ctx) =>
{
    var outcome = await ctx.UpdateWorkflowAsync(p!["wf_id"]!.GetValue<string>(), "echo", new { v = p?["probe"]?.GetValue<string>() }, waitMs: 10000);
    return await Report(ctx, "wfupdate", p, new()
    {
        ["status"] = outcome?["status"]?.DeepClone(), ["got"] = outcome?["result"]?["got"]?.DeepClone(),
    });
}),

Not in every language

• Crystal: Gateway-mediated update_workflow: Crystal sends updates with JoobQPro::Workflows::Client#update directly.

Tested against a real server in every language shown.

The answer is the outcome, {update_id, name, status, result, error, reason, message}: 200 with status completed, rejected or failed, or 202 with status: "accepted" when the handler did not answer within wait_ms (100 to 30000 ms, 10000 by default). Send it again with the same update_id (1 to 64 letters, digits, _ or -) to wait for the same update: a reused id answers the first outcome, so an update is applied once however often it is sent. Without one the server makes an id, and a retry is a new update.

Update, scope and change names are letters, digits, _ and - (stricter than step names); a workflow has at most 50 changes and 50 updates.

Change gates: change and gate

A change declares a point where the workflow's code changed (min defaults to 0, max to 1), and a gate on a node ({change, min?, max?}, open ended where a bound is absent; give at least one) puts the node on one side of it:

• a run's version of a change is fixed the first time it reaches a gated node, and recorded in its history. A new run takes the change's max. Only a run replayed onto the definition after it had already gone past that point gets 0 (the path from before the change), and only when the change's min is 0;
• a gated node runs only when that version is within the gate's bounds, and is skipped (reason version) otherwise.

A gate on an undeclared change, or one that admits neither max nor (with min 0) version 0, is refused (422): it would skip the node in every run.

Crystal

workflow "conf-gated-#{LANG}" do
  change "fmt", max: 2
  step "old_way", Ok, gate: JoobQPro::Workflows::VersionGate.new("fmt", max: 1)
  step "new_way", Ok, gate: JoobQPro::Workflows::VersionGate.new("fmt", min: 2)
end,

Python

wf("gated").change("fmt", max=2)
           .step("old_way", ok, gate=gate("fmt", max=1))
           .step("new_way", ok, gate=gate("fmt", min=2)),

TypeScript

wf('gated').change('fmt', { max: 2 })
  .step('old_way', ok, { gate: gate('fmt', { max: 1 }) })
  .step('new_way', ok, { gate: gate('fmt', { min: 2 }) }),

Ruby

wf("gated").change("fmt", max: 2)
           .step("old_way", OK, gate: gate("fmt", max: 1))
           .step("new_way", OK, gate: gate("fmt", min: 2)),

Go

wf("gated").
	Change("fmt", 0, 2).
	Step("old_way", ok, joobq.WithGate[*App](joobq.GateMax("fmt", 1))).
	Step("new_way", ok, joobq.WithGate[*App](joobq.GateMin("fmt", 2))),

Rust

wf("gated")
    .change("fmt", 0, 2)
    .step_with("old_way", ok, StepOptions { gate: Some(gate("fmt", None, Some(1))), ..Default::default() })
    .step_with("new_way", ok, StepOptions { gate: Some(gate("fmt", Some(2), None)), ..Default::default() }),

Java

wf("gated").change("fmt", 0, 2)
    .step("old_way", Worker::ok, StepOptions.<App>none().gate(DSL.gate("fmt", null, 1)))
    .step("new_way", Worker::ok, StepOptions.<App>none().gate(DSL.gate("fmt", 2, null))));

PHP

wf('gated')->change('fmt', max: 2)
    ->step('old_way', 'ok', gate: DSL::gate('fmt', max: 1))
    ->step('new_way', 'ok', gate: DSL::gate('fmt', min: 2)),

.NET

Wf("gated").Change("fmt", max: 2)
    .Step("old_way", Ok, gate: Dsl.Gate("fmt", max: 1))
    .Step("new_way", Ok, gate: Dsl.Gate("fmt", min: 2)),

Tested against a real server in every language shown.

fmt has max: 2, so every new run takes version 2: new_way (from version 2) runs and old_way (up to version 1) is skipped. old_way is the code from before the change, kept for runs that replay past it.

Each blueprint version is a definition of its own, and a run stays on the version it started on: runs started before you added the change keep the definition they started with, so they are not affected by the gate or by a new min.