ardiq 0.5.0

Documentation  •  Getting started  •  API reference  •  Benchmarks

A fast distributed task queue with a Rust core and a clean Python API, backed by Redis streams.

ArdiQ runs the worker loop and all Redis I/O in Rust (via PyO3 + tokio); you write tasks in plain Python. The two meet at a single async callback, with the GIL held only for the microseconds it takes to start a task and read its result — so a single process handles high concurrency.

Features

• 🦀 Rust core — the loop and Redis I/O run on tokio, off the GIL
• Priority queues — higher-priority tasks are consumed first
• Delayed & scheduled tasks (delay_ms / schedule_ms)
• Cron & recurring tasks (@app.cron) — 5-field cron (UTC) or every= intervals
• Automatic retries with quadratic backoff, configurable per task, or on demand (raise Retry)
• Enqueue by name (app.send("task", ...)) — producers never import the task module
• Error hooks (@app.on_error) — send every failed attempt to Sentry or your own reporter
• Typed failures (BrokerError) — catch "Redis is down" without a blanket except
• Unique task names, enforced at registration — a duplicate raises instead of silently shadowing
• Crash recovery — in-flight tasks of a dead worker are reclaimed (XAUTOCLAIM)
• Results with TTL, plus task status (queued / running / complete / not_found)
• Abort/cancel (job.abort()) — drops queued tasks and cancels running ones over pub/sub
• Sync & async tasks — blocking sync functions run in a thread pool
• CLI worker (ardiq run module:app) and burst mode (drain the queue and exit)

Performance

Because the worker loop and every Redis round-trip run in Rust — off the GIL — ArdiQ delivers near-top throughput at the lowest memory of any fast queue. Nothing in the suite beats it on throughput and memory at once.

Benchmarked head-to-head against arq, Taskiq, Streaq, Celery and Dramatiq on the same machine (1,000 tasks, one worker, 10 concurrent, 6 interleaved rounds):

• 🪶 Lightest of the fast queues — a third of Taskiq's memory, two thirds of Streaq's, at the same CPU throughput as Streaq.
• ⚡ Best work per megabyte, tied with arq — while running 9% faster than it on both workloads.
• 🎯 Predictable tail latency — p99 of 2,609 ms ±21 on CPU work, where the other async queues swing ±1,000 ms between rounds.
• 📈 Within 1.2% of the ceiling on I/O work — practically network-bound.

Not the fastest in raw throughput: Taskiq is 13% quicker on CPU-bound micro-tasks, and spends 91 MB doing it. ArdiQ pays about 0.3 ms per task to cross the Rust/Python boundary — 13% of a 2.4 ms task, 0.3% of a 100 ms one, which is why it wins on I/O and trades on CPU.

Throughput is shaped by hardware and workload, and the GIL caps in-process CPU work for every Python queue (ArdiQ included). The full, reproducible suite — with the honest caveats — lives in the benchmark repo, and the breakdown is in the performance guide.

When to use ArdiQ

Reach for ArdiQ when you want:

• High concurrency on a small footprint — async-native, with the loop and Redis I/O in Rust, so one process does a lot without eating memory.
• A modern, typed API — @app.task, awaitable enqueue, Job handles, results and status built in.
• Reliability out of the box — priorities, retries with backoff, delayed and scheduled tasks, and crash recovery via Redis consumer groups.
• Redis you already run — no extra broker to operate.

Consider the alternatives when:

• You need to saturate many CPU cores in one process — like every single-process Python queue, ArdiQ runs your task body under the GIL, so CPU-bound work is serial per worker (scale out with more workers). For heavy CPU fan-out, a prefork model (Celery, Dramatiq) can be simpler.
• You need a large, battle-tested ecosystem today — Celery has years of integrations, schedulers, and dashboards. ArdiQ is young and moving fast.
• You can't run Redis — ArdiQ is Redis-only by design.

ArdiQ sits alongside arq / Taskiq / Streaq as a modern async queue — its edge is the Rust core (memory and per-task overhead) and a batteries-included API.

Installation

$ pip install ardiq

That's everything — the library, the ardiq worker command, and a single runtime dependency (msgpack). Define tasks, enqueue them, and run a worker either from the CLI or from your own code (await app.run()).

You also need a Redis server — the quickest way is Docker:

$ docker run -d --name ardiq-redis -p 6379:6379 redis

or install it from your package manager (or redis.io).

Building from source (if you want to hack on ArdiQ itself): you'll need Rust and uv. Clone the repo and run uv sync.

Quickstart

Define an app and some tasks (example.py):

from ardiq import Ardiq

app = Ardiq(redis_url="redis://localhost:6379", queue_name="example")


@app.task()
async def add(a: int, b: int) -> int:
    return a + b


@app.task(max_retries=3)
def slow_double(x: int) -> int:   # sync task — runs in a thread
    return x * 2

Start a worker:

$ ardiq run example:app

Enqueue tasks from anywhere and read their results:

import asyncio
from example import add


async def main():
    job = await add.enqueue(2, 3)        # returns a Job handle
    print(job.id)
    print(await job.status())            # 'queued' | 'running' | 'complete'
    print(await job.result(timeout=5))   # waits → TaskResult(success=True, value=5, tries=1)


asyncio.run(main())

Or run the whole thing in one process with python example.py, which enqueues a few tasks and processes them in burst mode.

Enqueuing by name

The side that enqueues doesn't have to be the side that runs. app.send puts a task on the queue by name, so a web service can dispatch work without importing the task module — or its dependencies — at all:

from ardiq import Ardiq
from fastapi import FastAPI

api = FastAPI()
queue = Ardiq(redis_url="redis://localhost:6379", queue_name="example")


@api.post("/reports")
async def create_report(user_id: int):
    job = await queue.send("build_report", user_id, format="pdf")
    return {"job_id": job.id}

Nothing is checked locally: the name is resolved by the worker that picks the task up, and one it doesn't know fails there like any other error. For the enqueue options, app.ref hands back the same handle @app.task returns:

await queue.ref("build_report").options(delay_ms=60_000, priority="low").enqueue(7)

A ref can be enqueued but not called — there is no local function behind it.

Priority does not travel with the name. Everything else you put on @app.task(...) — max_retries, backoff_ms, timeout — is applied by the worker, which has the registry and can look it up. Priority is the exception: it picks which stream the task goes into, so it is settled by the producer, before the payload leaves. A task declared @app.task(priority="high") and dispatched by name lands in the app's default_priority instead, with no warning — pass it at the call site:

await queue.ref("build_report", priority="high").enqueue(7)

Since the fallback is the middle lane, forgetting it is survivable rather than disastrous, but the work still won't be where you declared it belongs.

Retries and error hooks

A task that raises is retried up to max_retries times, waiting tries² seconds between attempts (or the fixed backoff_ms you configure). Raise Retry to make that call from inside the task instead:

from ardiq import Retry


@app.task(max_retries=5)
async def call_api():
    response = await client.get(URL)
    if response.status_code == 429:
        raise Retry("rate limited", delay_ms=30_000)
    return response.json()

Retry still respects max_retries, so it can't loop forever; when the budget runs out the task fails with it as the error.

@app.on_error runs a hook on every failed attempt, before ArdiQ decides between retrying and failing — this is where a reporter like Sentry goes:

import sentry_sdk


@app.on_error
def report(ctx):
    sentry_sdk.capture_exception(ctx.exc)
    log.warning("%s failed on try %s (retrying: %s)", ctx.name, ctx.tries, ctx.will_retry)

The hook takes an ErrorContext(name, task_id, exc, tries, will_retry), may be sync or async, and can be registered more than once — all of them run. One that raises is logged and never changes the task's outcome.

It fires on timeouts, on every retry, and when a worker is handed a task it doesn't know. It does not fire on abort, nor for a Retry you raised yourself — only when that Retry finally gives up. Hooks run on the worker's event loop, so keep them quick.

When Redis is what failed — unreachable, refusing or dropping connections — the call raises BrokerError (→ ArdiqError → RuntimeError), so an enqueue in a request handler can be caught precisely instead of with a bare except RuntimeError:

from ardiq import BrokerError

try:
    job = await queue.send("build_report", user_id)
except BrokerError:
    raise HTTPException(503, "queue unavailable")

Shared resources (lifespan)

Tasks often need something expensive that should be built once per worker, not per task — a database pool, an HTTP client. @app.lifespan registers an async generator that sets up before the loop starts and tears down after it stops:

@app.lifespan
async def lifespan():
    pool = await asyncpg.create_pool(DSN)
    yield {"db": pool}          # entries land on app.state
    await pool.close()


@app.task()
async def count_users() -> int:
    return await app.state.db.fetchval("select count(*) from users")

Yield a mapping to populate app.state, or yield nothing and assign app.state.db = ... yourself. Either way app.state is available to async and sync tasks alike.

The hook only runs inside app.run(), so a process that just enqueues never opens the pool. Teardown runs even if the loop fails, and an exception during setup stops the worker before it takes any work.

Aborting tasks

job.abort() cancels a task whether it is waiting in the queue or already running on some worker:

job = await slow_report.options(delay_ms=60_000).enqueue()

if await job.abort():                # False if it already finished
    result = await job.result(timeout=5)
    print(result.aborted)            # True
    print(result.success)            # False

An aborted task ends as an ordinary failed TaskResult with aborted set, so it never retries and result(timeout=) returns as soon as it settles. What happens depends on where the task is when you call it:

Cancelling a running task needs a long-running worker: the worker subscribes to the queue's abort channel while it runs, which --burst skips. Aborts are still honored under burst, just not mid-flight.

Because cancellation is asyncio cancellation, a sync task can't be interrupted mid-call — the worker stops waiting on it and reports it aborted, but the thread runs to completion. Async tasks are cancelled at their next await, so a task that swallows CancelledError keeps going.

Recurring tasks

Register a task to run on a schedule with @app.cron — either a standard 5-field cron expression (evaluated in UTC) or a fixed every= interval:

@app.cron("0 3 * * *")            # daily at 03:00 UTC
async def nightly_report():
    ...


@app.cron(every=30)               # every 30s — int/float seconds or a timedelta
async def heartbeat():
    ...

Recurring tasks fire while a worker is running, and each occurrence is an ordinary task with its own result, status, retries and timeout. The cron syntax is the common subset — *, lists ,, ranges a-b, and steps */n — at minute resolution; use every= for sub-minute schedules.

Configuration

Ardiq(...) accepts:

@app.task(...) accepts name, max_retries (default 3), backoff_ms, timeout (seconds), and priority. @app.cron(spec, *, every=…, …) takes those same per-task options plus the schedule. Use task.options(delay_ms=…, schedule_ms=…, priority=…, task_id=…).enqueue(...) for one-off overrides.

Logging

ardiq run configures Python's logging for the process (INFO by default, DEBUG with --verbose/-v) and also initializes the Rust core's own logging at the same level. Worker lifecycle (worker starting, worker stopped) logs at INFO; task lifecycle logs at DEBUG (task started, task succeeded) through WARN (task retry scheduled) and ERROR (task failed, task unknown). Task args, kwargs, and results are never logged.

Logging inside a task is just standard logging — it works the same for async tasks and for sync tasks run via asyncio.to_thread:

import logging

logger = logging.getLogger(__name__)


@app.task()
async def send_email(to: str) -> None:
    logger.info("sending email to %s", to)
    ...

If you embed Ardiq outside the ardiq CLI, call logging.basicConfig(...) yourself (see example.py).

Development

$ docker compose up -d      # Redis on localhost:6379
$ uv run pytest             # test suite (needs Redis)
$ uv run ruff check .       # lint
$ uv run ty check ardiq tests   # type-check

After changing the Rust core, rebuild with uv sync --reinstall-package ardiq.

Contributing

Bug reports, docs fixes and features are all welcome — see CONTRIBUTING.md for setup, the layout of the codebase and what to open an issue about first. Questions and ideas go in Discussions.

License

MIT