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=…))).

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