How to implement a Celery-style chord

[anthonyatp]

Hi Taskiq team, thanks for the great project.

I’m using Taskiq as an async-friendly alternative to Celery and I’d love guidance on the idiomatic way to implement a Celery-style chord: trigger N tasks in parallel and then run a single callback only after all have finished.

Environment: RedisStreamBroker + RedisAsyncResultBackend (long TTL), FastAPI integration via taskiq_fastapi.init. I want clear behavior for retries, failures, and timeouts, and a straightforward way to aggregate results.
Here’s intent-only pseudo-code to illustrate the goal:

@broker.task
async def process_item(item_id: int) -> dict:
    ...

@broker.task
async def finalize() -> None:
    ...

async def orchestrate(items: list[int]) -> None:
    chord = (
        Chord(
            broker,
            process_item(i) for i in items,  # Create a task per item
        )
        .on_complete(finalize)
    )

    task = await chord.kiq()

I have looked into taskiq-pipelines but it's unclear how I would implement something like this. The new API design for pipeline looks promising but I was wondering if there was a more short-term solution you would recommend?

Thanks

[abderrahimchadli]

For a short-term solution, run the fan-out/fan-in coordinator in your FastAPI/service process rather than inside a worker task:

import asyncio

async def orchestrate(items: list[int]):
    handles = [await process_item.kiq(i) for i in items]
    results = await asyncio.gather(
        *(handle.wait_result(timeout=300) for handle in handles)
    )
    for result in results:
        result.raise_for_error()
    return await finalize.kiq([result.return_value for result in results])

Here finalize would need a parameter accepting the aggregated results. Submitting all child tasks first allows the available workers to process them concurrently; awaiting each result before submitting the next would serialize them. Keep keep_results=True and a result TTL longer than the whole workflow, including retries and downtime.

With the default retry middleware behavior (no_result_on_retry=True), retrying attempts aren't saved as final results. A final child error or a result-wait timeout prevents this coordinator from enqueueing the callback. A timeout does not cancel the other jobs.

This is not a durable chord: a service restart loses the coordinator, and a crash around callback enqueue can cause a missing or duplicate callback. For restart recovery, persist the group and child task IDs, record child completion idempotently, and use an atomic completion transition plus an outbox/idempotent finalizer. Don't keep a worker slot occupied waiting for its own child jobs—especially with limited concurrency.

API references: task handles, result errors, Redis result retention. This is an application-level pattern, not a claim that Taskiq provides Celery's chord guarantees.