Queries and client reads

ProQueries and client reads 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

Queries

A blueprint declares queries as references into the run's input and step outputs. The server answers them, by the engine's Query.run, so no worker is asked and a closed run still answers. A step that has not run answers null; an unknown query is 404. state and steps are built-in queries every run answers (and cannot be declared). Values with secret-looking names are masked. A query that needs code (a computed answer) is not supported.

Crystal

workflow "conf-query-#{LANG}" do
  step "a", Doubled, input: {x: input(:x)}
  wait_for "hold", signal: "release", timeout: 60.seconds
  step "b", Ok
  query "doubled", ref("steps.a.output.doubled")
  query "who", ref("input.who")
  query "released_by", ref("steps.hold.output.by")
end,

Python

wf("query").step("a", doubled, input={"x": input("x")}).wait_for("hold", signal="release", timeout_ms=60000)
           .step("b", ok)
           .query("doubled", output("a", "doubled")).query("who", input("who"))
           .query("released_by", output("hold", "by")),

TypeScript

wf('query')
  .step('a', async (i: any) => ({ doubled: i.x * 2 }), { input: { x: input('x') } })
  .waitFor('hold', { signal: 'release', timeoutMs: 60000 })
  .step('b', ok)
  .query('doubled', output('a', 'doubled'))
  .query('who', input('who'))
  .query('released_by', output('hold', 'by')),

Ruby

wf("query")
  .step("a", ->(i, _ctx) { { doubled: i["x"] * 2 } }, input: { x: input("x") })
  .wait_for("hold", signal: "release", timeout_ms: 60_000)
  .step("b", OK)
  .query("doubled", output("a", "doubled"))
  .query("who", input("who"))
  .query("released_by", output("hold", "by")),

Go

wf("query").
	Step("a", func(in map[string]any, ec *EC) (any, error) {
		return map[string]any{"doubled": num(in["x"]) * 2}, nil
	}, joobq.WithInput[*App](map[string]any{"x": joobq.Input("x")})).
	WaitFor("hold", "release", 60000).
	Step("b", ok).
	Query("doubled", joobq.Output("a", "doubled")).
	Query("who", joobq.Input("who")).
	Query("released_by", joobq.Output("hold", "by")),

Rust

wf("query")
    .step_with("a", |i: Value, _c: &Ctx| Ok(json!({"doubled": i["x"].as_i64().unwrap_or(0) * 2})),
        StepOptions { input: json!({"x": input(Some("x"))}), ..Default::default() })
    .wait_for("hold", "release", 60_000)
    .step("b", ok)
    .query("doubled", output("a", Some("doubled")))
    .query("who", input(Some("who")))
    .query("released_by", output("hold", Some("by"))),

Java

wf("query")
    .step("a", (i, ctx) -> map("doubled", num(m(i).get("x")) * 2), StepOptions.input(map("x", DSL.input("x"))))
    .waitFor("hold", "release", 60_000)
    .step("b", Worker::ok)
    .query("doubled", DSL.output("a", "doubled"))
    .query("who", DSL.input("who"))
    .query("released_by", DSL.output("hold", "by")),

PHP

wf('query')
    ->step('a', fn ($i, $ctx) => ['doubled' => $i['x'] * 2], input: ['x' => DSL::input('x')])
    ->waitFor('hold', 'release', 60000)
    ->step('b', fn () => ['ok' => true])
    ->query('doubled', DSL::output('a', 'doubled'))
    ->query('who', DSL::input('who'))
    ->query('released_by', DSL::output('hold', 'by')),

.NET

Wf("query")
    .Step("a", (i, c) => Task.FromResult<object?>(new { doubled = Num(i?["x"]) * 2 }), input: new { x = Dsl.Input("x") })
    .WaitFor("hold", "release", 60000)
    .Step("b", Ok)
    .Query("doubled", Dsl.Output("a", "doubled"))
    .Query("who", Dsl.Input("who"))
    .Query("released_by", Dsl.Output("hold", "by")),

Tested against a real server in every language shown.

Ask from the console (POST /api/v1/workflows/:id/query {name}, operator) or from code (query_workflow, below).

Client reads

A producer reads and annotates its runs through the gateway. Every route answers 404 for an unknown run and masks secret-looking payload fields like the console.

SDK call	Route	Answer
describe_workflow(id)	GET /sdk/workflows/:id	the run: {id, name, status, closed, run, started_at, closed_at, continued_to, output, failure, attributes, queries}
workflow_result(id, wait_ms)	GET /sdk/workflows/:id/result?wait_ms=	the same shape for the run at the end of the continue-as-new chain, after waiting up to wait_ms (0 to 25000) for it to close; closed: false means the wait ended first
workflow_history(id, after_seq, limit)	GET /sdk/workflows/:id/history?after_seq=&limit=	the events, oldest first ({type, data, step_id, attempt, ts, seq})
query_workflow(id, name)	POST /sdk/workflows/:id/query {name}	the query's answer
upsert_workflow_attributes(id, set, remove)	POST /sdk/workflows/:id/search-attributes {set?, remove?}	{id, attributes} of the open run that changed (409 when the chain is closed); the names must be registered

status is the engine's, in snake case: pending, running, completed, failed, terminated, cancelled, timed_out, continued_as_new; closed is true for every status but pending and running. Ids are sent as one URL path segment (percent-encoded).

Python

@job("conf.wfclient", queue=QUEUE)
def wfclient_job(payload, ctx):
    wf = payload["wf_id"]
    doubled = ctx.query_workflow(wf, "doubled")
    run = ctx.describe_workflow(wf)
    events = ctx.workflow_history(wf)
    changed = ctx.upsert_workflow_attributes(wf, set={"conf_tag": payload["probe"]})
    ctx.signal_workflow(wf, "release", {"by": payload["probe"]})
    result = ctx.workflow_result(wf, wait_ms=20000)
    report(ctx, "wfclient", payload, doubled=doubled, status=run.get("status"), closed=run.get("closed"),
           events=len(events), tag=(changed.get("attributes") or {}).get("conf_tag"),
           result_status=result.get("status"), result_closed=result.get("closed"))

TypeScript

defineJob<App>({ name: 'conf.wfclient', queue: QUEUE, perform: async (p, ctx) => {
  const wf = p.wf_id
  const doubled = await ctx.queryWorkflow(wf, 'doubled')
  const run = await ctx.describeWorkflow(wf)
  const events = await ctx.workflowHistory(wf)
  const changed = await ctx.upsertWorkflowAttributes(wf, { conf_tag: p.probe })
  await ctx.signalWorkflow(wf, 'release', { by: p.probe })
  const result = await ctx.workflowResult(wf, 20000)
  return report(ctx, 'wfclient', p, {
    doubled, status: run.status, closed: run.closed, events: events.length, tag: changed.attributes?.conf_tag,
    result_status: result.status, result_closed: result.closed,
  })
} }),

Ruby

class WfClientJob
  include JoobQ::Job
  job_name "conf.wfclient"
  queue QUEUE
  def self.perform(p, ctx)
    wf = p["wf_id"]
    doubled = ctx.query_workflow(wf, "doubled")
    run = ctx.describe_workflow(wf)
    events = ctx.workflow_history(wf)
    changed = ctx.upsert_workflow_attributes(wf, set: { conf_tag: p["probe"] })
    ctx.signal_workflow(id: wf, name: "release", payload: { by: p["probe"] })
    result = ctx.workflow_result(wf, wait_ms: 20_000)
    report(ctx, "wfclient", p, doubled: doubled, status: run["status"], closed: run["closed"],
                               events: events.size, tag: (changed["attributes"] || {})["conf_tag"],
                               result_status: result["status"], result_closed: result["closed"])
  end
end

Go

joobq.DefineJob[*App]("conf.wfclient", opts, func(p map[string]any, ec *EC) (any, error) {
	wf, _ := p["wf_id"].(string)
	doubled, err := ec.QueryWorkflow(wf, "doubled")
	if err != nil {
		return nil, err
	}
	run, err := ec.DescribeWorkflow(wf)
	if err != nil {
		return nil, err
	}
	events, err := ec.WorkflowHistory(wf, 0, 200)
	if err != nil {
		return nil, err
	}
	changed, err := ec.UpsertWorkflowAttributes(wf, map[string]any{"conf_tag": p["probe"]}, nil)
	if err != nil {
		return nil, err
	}
	if err := ec.SignalWorkflow(wf, "release", map[string]any{"by": p["probe"]}); err != nil {
		return nil, err
	}
	result, err := ec.WorkflowResult(wf, 20000)
	if err != nil {
		return nil, err
	}
	attrs, _ := changed["attributes"].(map[string]any)
	return report(ec, "wfclient", p, map[string]any{
		"doubled": doubled, "status": run["status"], "closed": run["closed"], "events": len(events),
		"tag": attrs["conf_tag"], "result_status": result["status"], "result_closed": result["closed"],
	})
}),

Rust

JobDef::new("conf.wfclient", queue.clone(), |p: Value, ctx: &Ctx| {
    let wf = p["wf_id"].as_str().unwrap_or("").to_string();
    let doubled = ctx.query_workflow(&wf, "doubled")?;
    let run = ctx.describe_workflow(&wf)?;
    let events = ctx.workflow_history(&wf, 0, 200)?;
    let mut set = joobq::Map::new();
    set.insert("conf_tag".into(), p["probe"].clone());
    let changed = ctx.upsert_workflow_attributes(&wf, Some(set), None)?;
    ctx.signal_workflow(&wf, "release", json!({"by": p["probe"]}))?;
    let result = ctx.workflow_result(&wf, 20_000)?;
    report(ctx, "wfclient", &p, json!({
        "doubled": doubled, "status": run["status"], "closed": run["closed"], "events": events.len(),
        "tag": changed["attributes"]["conf_tag"], "result_status": result["status"], "result_closed": result["closed"],
    }))
}),

Java

JobDefinition.<App>of("conf.wfclient", QUEUE, 3, (p, ctx) -> {
    String wfId = (String) m(p).get("wf_id");
    Object doubled = ctx.queryWorkflow(wfId, "doubled");
    Map<String, Object> run = ctx.describeWorkflow(wfId);
    List<Map<String, Object>> events = ctx.workflowHistory(wfId);
    Map<String, Object> changed = ctx.upsertWorkflowAttributes(wfId, map("conf_tag", m(p).get("probe")), null);
    ctx.signalWorkflow(wfId, "release", map("by", m(p).get("probe")));
    Map<String, Object> result = ctx.workflowResult(wfId, 20_000);
    Object attrs = changed.get("attributes");
    return report(ctx, "wfclient", p, map("doubled", doubled, "status", run.get("status"), "closed", run.get("closed"),
        "events", events.size(), "tag", attrs instanceof Map<?, ?> a ? a.get("conf_tag") : null,
        "result_status", result.get("status"), "result_closed", result.get("closed")));
}),

PHP

final class WfClientJob extends ConfJob
{
    public static function jobName(): string { return 'conf.wfclient'; }
    public static function perform(array $p, ExecutionContext $ctx): mixed
    {
        $wf = $p['wf_id'];
        $doubled = $ctx->queryWorkflow($wf, 'doubled');
        $run = $ctx->describeWorkflow($wf);
        $events = $ctx->workflowHistory($wf);
        $changed = $ctx->upsertWorkflowAttributes($wf, set: ['conf_tag' => $p['probe']]);
        $ctx->signalWorkflow($wf, 'release', ['by' => $p['probe']]);
        $result = $ctx->workflowResult($wf, waitMs: 20000);
        return report($ctx, 'wfclient', $p, [
            'doubled' => $doubled,
            'status' => $run['status'] ?? null,
            'closed' => $run['closed'] ?? null,
            'events' => count($events),
            'tag' => $changed['attributes']['conf_tag'] ?? null,
            'result_status' => $result['status'] ?? null,
            'result_closed' => $result['closed'] ?? null,
        ]);
    }
}

.NET

JobDefinition<App>.Define("conf.wfclient", Queue, 3, async (p, ctx) =>
{
    var probe = p?["probe"]?.GetValue<string>();
    var wf = p!["wf_id"]!.GetValue<string>();
    var doubled = await ctx.QueryWorkflowAsync(wf, "doubled");
    var run = await ctx.DescribeWorkflowAsync(wf);
    var events = await ctx.WorkflowHistoryAsync(wf);
    var changed = await ctx.UpsertWorkflowAttributesAsync(wf, set: new() { ["conf_tag"] = probe });
    await ctx.SignalWorkflowAsync(wf, "release", new { by = probe });
    var result = await ctx.WorkflowResultAsync(wf, waitMs: 20000);
    return await Report(ctx, "wfclient", p, new()
    {
        ["doubled"] = doubled, ["status"] = run?["status"]?.DeepClone(), ["closed"] = run?["closed"]?.DeepClone(),
        ["events"] = events.Count, ["tag"] = changed?["attributes"]?["conf_tag"]?.DeepClone(),
        ["result_status"] = result?["status"]?.DeepClone(), ["result_closed"] = result?["closed"]?.DeepClone(),
    });
}),

Not in every language

• Crystal: Gateway-mediated workflow client reads: Crystal uses JoobQPro::Workflows::Client directly.

Tested against a real server in every language shown.

Without a Pro key every one of these routes answers 402 pro_required.