Canvas Primitives

chain, group, and chord are methods on the same workflow builder you use for everything else — they just add .step() calls with the after dependency wiring done for you. The result is a normal DAG workflow you .submit() like any other.

Canvas.chain, Canvas.group, and Canvas.chord build a Workflow from a few Canvas.link(name, task, payload) steps — the same DAG a hand-written builder chain would produce. Submit the result like any other workflow, and keep adding steps to it with the regular builder if you need more.

Coming from Celery

flexiq's canvas mirrors Celery's chain / group / chord / .s() / .si() / chunks / starmap, with a few deliberate differences:

Every shape below, and the call that replaces it:

Two limits shape all of them, and neither is a gap you can route around. depends_on orders jobs but passes nothing between them — every call carries the arguments it was built with. And a result is not readable from here: job.result is an Option<Vec<u8>> with no typed reader exposed. Where a shape wants a value to flow, the producing task has to write it somewhere the consumer can read, and the consumer gets the key.

Each step's return value flows automatically into the next via .s(). Use .si() when a step should not receive the previous result:

chain(
    step_a.s(input_data),
    step_b.si(independent_data),
    step_c.s(),
).apply(queue)

Pass { after } to chain a canvas block onto an existing step.

Building the chain link by link means the producer has to be enqueued before its successor — you need extracted.id to write the next call. That is a plain data dependency, not a ceremony, and it falls out of the code:

let mut previous: Option<String> = None;
for url in urls {
    let call = fetch::call(url);
    let call = match &previous {
        Some(id) => call.depends_on([id]),
        None => call,
    };
    previous = Some(queue.enqueue(call)?.id);
}

Storage checks a dependency at enqueue time, so naming an id that does not exist, one that is already Dead or Cancelled, or one in another namespace fails the enqueue call outright rather than producing a job that can never run.

A step that merely fails is different: it stays in the table, its dependents stay Pending, and the chain simply stops. Nothing propagates the failure downstream and nothing cancels the tail — watch for stalled work instead of waiting for an error.

Concurrency limits

jobs = group(
    *[fetch.s(url) for url in urls],
    max_concurrency=5,
).apply(queue)

Canvas.group always starts group members as roots. To hang a parallel group off an existing step, add the steps with the builder instead — Canvas shortcuts always start from roots:

Workflow wf = Workflow.named("notify")
        .step("prepare", prepareTask, 1)
        .step("email", sendEmail, message, "prepare")
        .step("sms", sendSms, message, "prepare");
// prepare → (email | sms)

A group has no root/non-root distinction to work around — depends_on is an option on each call, so to hang one off an earlier step, give every member the same predecessor:

let prepare = queue.enqueue(prepare::call(order_id))?;

let fanned = queue.enqueue_batch(vec![
    notify::call(order_id, "email".to_string()).depends_on([&prepare.id]),
    notify::call(order_id, "sms".to_string()).depends_on([&prepare.id]),
])?;
// prepare → (email | sms)

Three different tasks would be three enqueue calls instead, since a batch is generic over one T: Task. They still run in parallel; the batch buys atomicity of the insert, not the parallelism.

There is no max_concurrency on a group either, because the cap has nothing to attach to — nothing records that these jobs are one group. Declare it on the task and it holds across every caller:

#[flexiq::task(max_concurrent = 5)]
fn fetch(url: String) -> flexiq::Outcome<()> { /* … */ Ok(()) }

max_concurrent is a cluster-wide ceiling on concurrent runs of that task. max_in_flight_per_task is the per-worker sibling, and rate_limit = "100/m" caps starts per unit of time rather than overlap.

Because the callback cannot read what the members produced, the aggregation has to happen somewhere both can reach. A run id both sides derive from is usually enough:

let members = queue.enqueue_batch(vec![
    query_region::call(run_id.clone(), "east".to_string()),
    query_region::call(run_id.clone(), "west".to_string()),
])?;

// Each member writes under `run_id`; `merge` reads them back by the same key.
let merge = queue.enqueue(
    merge::call(run_id).depends_on(members.iter().map(|job| job.id.as_str())),
)?;

depends_on takes anything iterating into String, which is why &job.id and job.id.as_str() both work.

Saga compensation

Canvas primitives support in-thread saga compensation via .with_compensation(). When a step fails, flexiq dispatches the compensators for the steps that already succeeded — no workflow engine required.

chain compensation

Attach one compensator per chain step. None skips a slot. On failure, compensators for completed steps run in reverse order:

from flexiq import chain

@queue.task()
def step_a(x: int) -> int:
    return charge_payment(x)

@queue.task()
def step_b(x: int) -> str:
    return create_order(x)

@queue.task()
def refund(forward_args, forward_kwargs, forward_result):
    payment_id = forward_result
    payment_api.refund(payment_id)

@queue.task()
def cancel_order(forward_args, forward_kwargs, forward_result):
    order_id = forward_result
    order_api.cancel(order_id)

result = chain(
    step_a.s(100),
    step_b.s(),
).with_compensation([refund, cancel_order]).apply(queue)

with_compensation requires exactly one entry per chain step (a slot can be None to skip that step). Only steps that completed are compensated, in reverse order, one at a time — the failed step itself never ran to completion, so it has nothing to undo. If step_b fails after step_a succeeded, only refund (step_a's compensator) is enqueued; cancel_order would run only in a longer chain where step_b completed and a later step failed.

group compensation

One compensator per group member. On failure of any member, compensators for the succeeded members dispatch in parallel. The failed member is not compensated:

from flexiq import group

g = group(
    debit_account.s(100),
    reserve_inventory.s("item-7"),
    notify_warehouse.s(42),
).with_compensation([
    refund_account,
    release_inventory,
    cancel_notification,
])

g.apply(queue)

chord compensation

chord.with_compensation takes separate group= and callback= arguments:

from flexiq import chord, group

ch = chord(
    group(fetch.s("url1"), fetch.s("url2")),
    merge.s(),
).with_compensation(
    group=[rollback_fetch1, rollback_fetch2],
    callback=rollback_merge,
)

ch.apply(queue)

Compensation order:

• Group member fails before the callback runs → group compensators for the succeeded members dispatch in parallel.
• Callback fails after the group succeeded → group compensators dispatch in parallel, same as above.

Compensator contract

Every compensator receives three positional args — the same contract as DAG-level saga compensators:

@queue.task()
def my_compensator(forward_args: tuple, forward_kwargs: dict, forward_result: object) -> None:
    # forward_result is the return value of the step being compensated.
    ...

Idempotency

Each compensator is enqueued with a deterministic unique key of the form canvas_compensation:{run_id}:{slot}, where run_id is a fresh UUID generated when apply() starts. The dedup guarantee is per-run: it stops one apply() call's failure handling from double-dispatching the same compensator slot. It does not carry across separate apply() calls — each call gets its own run_id, so retrying a whole canvas from scratch (e.g. calling apply() again after a crash) re-dispatches every compensator.

Passing None to disable compensation

# Disables all compensation (equivalent to not calling with_compensation).
chain(a.si(), b.si()).with_compensation(None).apply(queue)

# Disables compensation for one specific slot.
chain(a.si(), b.si()).with_compensation([compensate_a, None]).apply(queue)

Saga compensation

Canvas steps are ordinary workflow steps, so a CanvasStep accepts the same compensate option as .step():

.chain([
  { name: "reserve", task: "reserveInventory", compensate: "unreserveInventory" },
  { name: "charge", task: "chargePayment", compensate: "refundPayment" },
])

Rollback then follows the regular DAG-level saga rules — reverse-dependency order, driven by the workflow run's own state machine, not a canvas-specific mechanism. See Saga compensation.

Saga compensation

Canvas.link() only takes a name, task, and payload — there's no compensate option on the shorthand itself. For rollback on a canvas-built step, mix in the regular builder: Canvas.chain(...) (and group/chord) return a plain Workflow, and Workflow.named(...).step(...) exposes a .compensate(...) builder method for saga compensation.

Saga compensation

There is no compensate option — not on EnqueueOptions, not on #[flexiq::task], and nothing walks a failure back down a depends_on chain. Enqueued jobs are independent rows; one failing leaves the others exactly where they are.

Rollback lives inside a single task instead, written over durable steps so the undo knows what was actually done:

#[flexiq::task]
fn checkout(order_id: String) -> flexiq::Outcome<()> {
    let mut step = flexiq::current_step();

    let hold = step.run_keyed("reserve", &order_id, || reserve(&order_id))?;

    // The charge's failure is this step's *value*, not an abort — see below.
    let charged: Result<Charge, String> = step.run_keyed("charge", &order_id, || {
        Ok(charge(&order_id).map_err(|e| e.to_string()))
    })?;

    if let Err(reason) = charged {
        step.run_keyed("unreserve", &order_id, || unreserve(&hold))?;
        return Err(flexiq::TaskError::fatal(reason).into());
    }
    Ok(())
}

Two things make that work, and neither is optional.

run_keyed, not run: an unkeyed step is matched by its position in the recorded sequence, so a compensating branch asks for a step the earlier attempt never recorded there and the run fails with a divergence instead of undoing anything. A keyed step is matched by key wherever it sits.

And the forward failure is committed as a value rather than raised. A step whose body aborts commits nothing and leaves its position outstanding, and the session refuses to start another step while one is — so the compensator would fail with started while step 'charge' is still uncommitted rather than run.

Saga compensation covers the shape in full.

Node has no chunks / starmap helper — both are just a group of pre-sized steps. Build the batches in JS and hand them to .group():

// chunks: split items into batches of 100
const size = 100;
const batches = Array.from({ length: Math.ceil(items.length / size) }, (_, i) =>
  items.slice(i * size, (i + 1) * size),
);
queue.workflows
  .define("batch")
  .group(batches.map((b, i) => ({ name: `chunk-${i}`, task: "processBatch", args: [b] })))
  .submit();

// starmap: one step per arg tuple
queue.workflows
  .define("pairs")
  .group([[1, 2], [3, 4]].map(([a, b], i) => ({ name: `add-${i}`, task: "add", args: [a, b] })))
  .submit();

For dynamic batching driven by an upstream result, use fanOut / fanIn.

Java has no chunks / starmap shortcut — they are a group of fixed steps. Slice in Java and pass the links to Canvas.group:

List<Canvas.Link> links = new ArrayList<>();
for (int i = 0; i < items.size(); i += 100) {
    List<Item> batch = items.subList(i, Math.min(i + 100, items.size()));
    links.add(Canvas.link("chunk-" + (i / 100), processBatch, batch));
}
Workflow wf = Canvas.group("batch", links.toArray(new Canvas.Link[0]));
queue.submitWorkflow(wf);

For dynamic per-item fan-out, use Step.Builder.fanOut(...) / .fanIn(...).

Neither exists, and neither needs to — both are a Vec<TaskCall<T>> you build before handing it over:

// chunks: one call per slice of 100
let calls: Vec<_> = items
    .chunks(100)
    .map(|batch| process_batch::call(batch.to_vec()))
    .collect();
let jobs = queue.enqueue_batch(calls)?;

// starmap: one call per tuple, unpacked by the call signature itself
let jobs = queue.enqueue_batch(
    vec![(1, 2), (3, 4), (5, 6)]
        .into_iter()
        .map(|(a, b)| add::call(a, b))
        .collect(),
)?;

Starmap's unpacking is free here: add::call(a, b) takes the task's own parameters, so a tuple arity that does not match is a compile error rather than a runtime one. Chunking is just slice::chunks.

Dynamic batching — where the batch size comes from an upstream result — has no equivalent. Enqueue the follow-up work from inside the task that computed it, using a FlexiQ handle the process already holds.

Use canvas for quick one-off pipelines. Use DAG workflows for production pipelines that need monitoring, conditions, or complex topologies.

Canvas is not a separate engine in Node — chain / group / chord are methods on WorkflowBuilder that produce an ordinary DAG workflow. Reach for conditions, gates, or sub-workflows on the same builder when you need them; see Workflows.

Canvas.chain / group / chord return a plain Workflow you can keep extending with the builder (conditions, gates, sub-workflows) — canvas and DAG are the same object. See Workflows.

The choice is between depends_on on plain jobs and the flexiq-workflows crate, which you opt into with features = ["workflows"]:

lifecycle::submit_workflow is the entry point, and it is deliberately low-level: it takes the serialized graph, a step_metadata entry and a pre-encoded payload for every node that gets a job, and refuses the whole call if one is missing rather than leaving half a run behind. It returns a WorkflowRunHandle — run_id and definition_id — which is what you poll on.

Reach for it when you want the run rows. For a pipeline you only need to order, depends_on costs nothing and adds no crate.