Canvas Primitives

chain / group / chord — shorthand composition for common task pipeline and workflow shapes.

flexiq's canvas gives you three shorthand primitives for common pipeline shapes: chain (sequential), group (parallel), and chord (parallel with a callback) — composing task calls without hand-wiring dependencies yourself.

Not the DAG workflow engine

Canvas primitives are in-thread orchestrators, independent of DAG workflows — no persistent run row in storage, no workflow engine required. chain enqueues each step and blocks on its result before dispatching the next; group enqueues every member in parallel; chord runs a group then a callback. See "Canvas vs DAG workflows" below for when to reach for each.

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:

Celery	flexiq	Difference
`chain(...).apply_async()`	`chain(...).apply(queue)`	`.apply()` here submits the canvas; pass the `queue`.
`chain(...).apply()` (eager)	—	flexiq has no eager `.apply()`; it always enqueues for workers.
`AsyncResult.get(timeout=)`	`job.result(timeout=)`	Same idea, different name.
`group(...)` → `GroupResult`	`group(...).apply(queue)` → `list[JobResult]`	flexiq returns a plain list of handles — iterate and call `.result()` on each.
`.s()` / `.si()`	`.s()` / `.si()`	Identical (mutable vs immutable signature).

`.apply()` means submit, not run

The biggest gotcha: in Celery, .apply() executes the canvas locally and synchronously. In flexiq, .apply(queue) enqueues it for workers, and you await results with .result(). There is no eager local execution.

Building blocks

Each primitive is composed from small step descriptors:

A Signature wraps a task call for deferred execution:

python

from flexiq import chain, group, chord

sig = add.s(1, 2)    # Mutable — receives previous result as first arg
sig = add.si(1, 2)   # Immutable — ignores previous result

A CanvasStep names a step and its task; it accepts every .step() option except after, which the shorthand manages for you:

{ name: "extract", task: "extract" }
{ name: "load", task: "load", maxRetries: 5 }

Canvas.link(name, task, payload) binds a step name to a registered task and its payload:

java

Canvas.link("extract", extractTask, 1)
Canvas.link("transform", transformTask, 2)

Chain

Run steps in sequence:

python

result = chain(
    extract.s("https://api.example.com/users"),
    transform.s(),
    load.s(),
).apply(queue)

print(result.result(timeout=30))

queue.workflows
  .define("etl")
  .chain([
    { name: "extract", task: "extract" },
    { name: "transform", task: "transform" },
    { name: "load", task: "load", maxRetries: 5 },
  ])
  .submit();
// extract → transform → load

java

Workflow etl = Canvas.chain("etl",
        Canvas.link("extract", extractTask, 1),
        Canvas.link("transform", transformTask, 2),
        Canvas.link("load", loadTask, 3));

queue.submitWorkflow(etl);
// extract → transform → load

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

python

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

A chain step is an ordinary workflow step wired with after — it does not automatically receive its predecessor's return value, and args are fixed at build time. To act on an upstream result, aggregate it with fan-out / fan-in instead of a plain chain step.

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

Like any workflow step, a chain link does not automatically receive its predecessor's return value — pass the payload explicitly, or read upstream results via the workflow context.

Group

Run steps in parallel:

python

jobs = group(
    process.s(1),
    process.s(2),
    process.s(3),
).apply(queue)

results = [j.result(timeout=30) for j in jobs]

queue.workflows
  .define("notify")
  .step("prepare", "prepare")
  .group(
    [
      { name: "email", task: "sendEmail" },
      { name: "sms", task: "sendSms" },
      { name: "push", task: "sendPush" },
    ],
    { after: "prepare" },
  )
  .submit();
// prepare → (email | sms | push)

java

Workflow notify = Canvas.group("notify",
        Canvas.link("email", sendEmail, message),
        Canvas.link("sms", sendSms, message),
        Canvas.link("push", sendPush, message));

queue.submitWorkflow(notify);
// email | sms | push

Concurrency limits

python

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:

java

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

Chord

Fan-out with a callback — run tasks in parallel, then run a final task once every member finishes (whether the callback sees the members' results is SDK-specific — see the notes below):

python

result = chord(
    group(
        fetch.s("https://api1.example.com"),
        fetch.s("https://api2.example.com"),
        fetch.s("https://api3.example.com"),
    ),
    merge.s(),
).apply(queue)

queue.workflows
  .define("report")
  .chord(
    [
      { name: "q1", task: "queryRegion" },
      { name: "q2", task: "queryRegion" },
    ],
    { name: "merge", task: "merge" },
  )
  .submit();
// (q1 | q2) → merge

java

Workflow report = Canvas.chord("report",
        Canvas.link("merge", mergeTask, 0),      // callback
        Canvas.link("q1", queryRegion, "east"),  // group...
        Canvas.link("q2", queryRegion, "west"));

queue.submitWorkflow(report);
// (q1 | q2) → merge

The callback receives the list of member results as its first argument — merge above is called with [result1, result2, result3]. Use an immutable signature (merge.si()) to opt out.

chord's callback runs after the group but receives its own args — the members' results are not auto-passed. To aggregate results into one step, use fan-out / fan-in, which collects child results into the combiner's argument.

A chord's callback runs after the group but receives its own payload — the members' results are not auto-passed. To aggregate results into one step, use fan-out / fan-in, which collects child results into the combiner's payload.

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.

Canvas compensation is in-thread: compensators are enqueued synchronously during the apply() call that raised. For DAG-level sagas with full observability and sub-workflow propagation, see the Sagas guide.

chain compensation

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

python

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:

python

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:

python

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.

callback='s compensator is accepted by with_compensation() but is not currently dispatched on callback failure — only the group members are ever rolled back. Handle callback-side rollback inside the callback task itself until this gap closes.

Compensator contract

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

python

@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

python

# 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.

chunks / starmap

python

from flexiq import chunks, starmap

# Batch processing — split 1000 items into groups of 100
results = chunks(process_batch, items, chunk_size=100).apply(queue)

# Map-reduce pattern
result = chord(
    chunks(process_batch, items, chunk_size=100),
    merge_results.s(),
).apply(queue)

# Tuple unpacking
results = starmap(add, [(1, 2), (3, 4), (5, 6)]).apply(queue)

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:

java

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(...).

Canvas vs DAG workflows

Feature	Canvas	DAG Workflows
Setup	No imports needed	`from flexiq.workflows import Workflow`
Topology	Linear chains, flat groups	Arbitrary DAGs
Fan-out	Static (known at build time)	Dynamic (from return values)
Conditions	None	`on_success`, `on_failure`, `always`, callables
Error handling	Per-task retries only	Workflow-level strategies
Saga compensation	`chain` / `group` / `chord`	DAG-level, with sub-workflow propagation
Approval gates	No	Yes
Sub-workflows	No	Yes
Incremental runs	No	Yes
Status tracking	Per-job only	Per-workflow + per-node
Visualization	No	Mermaid / DOT

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.