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 / retry_delay are stored but not applied, #42), 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.
  • local broker: per host. The counter lives in the shared fp-brokerd, so it holds across processes on one machine (the local broker is single-host anyway).
  • memory broker: per process.
  • rabbitmq broker: per node. The running count is local, so N nodes can run up to N × 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=…))).

Losing a lock: wait, decline, or park it (on_lock_conflict)

A topic limit throttles; a lock excludes. When a task asks for one and somebody else holds it, there are three defensible answers, and which one is right depends entirely on what the message means:

policy what happens use it when
requeue back on the queue, invisible for a jittered backoff, retried after the message is distinct work — dropping it loses it
reject ACKed and dropped, recorded as REJECTED the message is a redundant duplicate of work already running
dead_letter parked in the DLQ, recorded as REJECTED you want the lost work kept for inspection or replay

The default is inferred from how you asked for the lock, because the two ways do not mean the same thing:

  • allow_concurrent=False → reject. "Never two at once" makes a second copy redundant. Queuing duplicates instead builds a backlog that stampedes the moment the lock frees — the classic cron-overrun pile-up.
  • an explicit lock_name → requeue. "Serialise these work items" means each message is its own work; dropping it loses it.

Override per task, or fleet-wide with the on_lock_conflict setting (a task's own choice wins):

@TaskWorker.register(name="fetch.profile", allow_concurrent=False)
def fetch_profile(): ...            # duplicate runs are declined

TaskWorker.submit(
    process_item,
    extra_context={"lock_name": f"item:{item_id}", "on_lock_conflict": "requeue"},
)

On memory, local and postgres the wait happens in the broker (a visibility timeout, #20), so the worker returns its pool slot immediately. On rabbitmq, which cannot delay a requeue, the worker waits instead — that slot is unavailable meanwhile, so a heavily contended lock there eats worker capacity that other topics need. See the broker matrix.

⚠️ dead_letter is for locks that are rarely contended. The DLQ exists for poison messages; routing routine contention into it floods it. The retry loop runs at roughly one cycle per second per worker, which is thousands of rows an hour on a fleet.

A rejection is never silent — it emits TaskStatus.REJECTED, distinct from SKIPPED (overloaded) and from FAILED (nothing failed). That visibility is the point: the original behaviour dropped every contended message with no record, so a queue given a global lock quietly drained itself.

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", topic="joins")
def join(parent_task_id, expected, attempt=0):
    children = get_children_statuses(parent_task_id)  # query TaskContextDB
    if len(children) < expected or not all(s in TERMINAL for s in children):
        if attempt >= 240:          # ~1 h at 15 s: give up, proceed with what completed
            ...
            return
        # Not done yet: wait, then re-queue myself.
        time.sleep(15)
        TaskWorker.submit(join, kwargs={"parent_task_id": parent_task_id,
                                        "expected": expected, "attempt": attempt + 1})
        return
    ...  # all fan-out tasks are terminal — proceed

There is no delayed submit(): retry_delay= is stored on the context but not applied (#42), so passing it re-queues immediately and the join spins as fast as the broker answers. The time.sleep() holds a worker slot while it waits, which is why the example puts joins on their own topic.

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.

It is not crash-safe by default. A child whose worker died mid-task keeps a non-terminal report. The watchdog.cleanup_stuck_tasks task, which flips such reports to DEAD, is registered but not scheduled: nothing runs it unless you add a schedule for it. The orphan reaper (docs/broker/orphan-recovery.md) requeues the broker message but does not finalise the old report. That is why the attempt bound above matters: a permanently stuck fan-out degrades to "proceed with what completed" instead of looping forever.

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.