Skip to content

Task orchestration: chaining, fan-out & completion

tasks_worker has no dedicated "workflow" / DAG engine, but three primitives compose into one — parent/child task trees, per-topic concurrency limits, and queryable task status. This doc shows how to build a multi-step pipeline, a serialized fan-out, and a completion join on top of them.

These are patterns over existing primitives, not a separate API. There is (as of 0.3.x) no built-in after=[…] / "fire when these all finish" helper — the join below is a hand-rolled poll. A first-class join primitive is a reasonable future addition; until then, use the pattern here.

Parent/child task trees

parent_task_id is more than retry linking — it is the structural link that makes a set of tasks one pipeline. When submit() is called from inside a running task, it reads the ambient current_task_ctx and sets the new task's parent_task_id to the caller automatically:

# Inside a running task, this call needs NO explicit parent_task_id —
# the child inherits the current task as its parent.
TaskWorker.submit(next_step, kwargs={"share_id": share_id})

The whole tree renders under the parent in the tasks UI, and every child is reachable by querying TaskContextDB.parent_task_id == <parent_task_id>. Pass parent_task_id= explicitly only for the head of a pipeline (submitted from outside any task) if you want to attach it to an existing root.

Chaining (a sequential pipeline)

Each step submits its successor on its own success — so reaching a step is proof every prior step succeeded, and the terminal step running is the pipeline's completion signal:

@TaskWorker.register(name="pipe.fetch")
def fetch(share_id): ...; TaskWorker.submit(enrich, kwargs={"share_id": share_id})

@TaskWorker.register(name="pipe.enrich")
def enrich(share_id): ...; TaskWorker.submit(finalize, kwargs={"share_id": share_id})

@TaskWorker.register(name="pipe.finalize")
def finalize(share_id): ...   # terminal — nothing left to submit

Make each step idempotent (re-running is a no-op when its output already exists): tasks_worker does not auto-retry (max_retries defaults to 0), so recovery is "re-submit the step", and a chain that dies mid-way is resumed by re-submitting from the failed step. Keep durable pipeline state (which step a given entity reached) in your own domain table, not inferred from task rows, which may be pruned.

Serialized fan-out (concurrency-limited topic)

To fan a step into N per-item tasks (e.g. one OCR task per image in a carousel) without hammering a downstream service, put those tasks on a dedicated topic with a concurrency limit. Declare the topic on the task, and set the limit at startup:

@TaskWorker.register(name="pipe.ocr_image", topic="ocr")
def ocr_image(share_id, asset_id): ...

# app startup, before uvicorn:
TaskWorker.setup_broker(topic_settings={
    "ocr": {"global_concurrency": 1},   # or an int directly: {"ocr": 1}
})

# fan out — all N queue immediately, but are consumed one at a time:
for asset in assets:
    TaskWorker.submit(ocr_image, kwargs={"share_id": s, "asset_id": asset.id})

Enforcement is broker-dependent (see broker matrix):

  • postgres broker — cluster-wide & atomic. The claim query only hands out a message when running-count-on-topic < concurrency_limit within the same claim, guarding the multi-worker race. Use this broker when you need a real, shared concurrency ceiling.
  • memory / local brokers — per-node. The counter lives in-process, so two nodes can each run up to the limit.

The limit can also be changed at runtime via broker.set_topic_concurrency_limit(topic, limit) (or set_topic_config(topic, TopicConfig(concurrency_limit=…))).

Completion join (wait for a fan-out to finish)

Since fan-out children inherit the parent's parent_task_id, "are they all done?" is a status query over the parent's children. Gate the next step on every child being in a terminal TaskStatus:

from fastpluggy_plugin.tasks_worker.core.status import TaskStatus

TERMINAL = {
    TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED,
    TaskStatus.CANCELLED, TaskStatus.MANUAL_CANCELLED,
    TaskStatus.ERROR, TaskStatus.TIMEOUT, TaskStatus.DEAD,
}

@TaskWorker.register(name="pipe.join")
def join(parent_task_id, expected):
    children = get_children_statuses(parent_task_id)  # query TaskContextDB
    if len(children) < expected or not all(s in TERMINAL for s in children):
        # not done yet — re-queue myself and check again shortly
        TaskWorker.submit(join, kwargs={"parent_task_id": parent_task_id,
                                        "expected": expected}, retry_delay=15)
        return
    ...  # all fan-out tasks are terminal — proceed

Why gate on task status rather than on the fan-out's output (rows written, files produced)? Because status is generic (no per-job "expected output" predicate) and, critically, crash-safe: the built-in watchdog (watchdog.cleanup_stuck_tasks) flips a task whose worker pid/thread has vanished to DEAD — a terminal state — so a crashed child cannot wedge the join forever. Bound the poll with a max attempt count so a permanently-stuck fan-out degrades to "proceed with what completed" instead of looping.

When to reach for a full sequential chain instead

If the fan-out items are few and cheap, a plain sequential loop inside one task (no fan-out, no join) is simpler and needs none of the above — prefer it until per-item parallelism or per-item retry/observability actually pays for the join's complexity.