ardiq 0.1.1

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

ArdiQ

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.

Early stage. The engine is solid — priorities, retries with backoff, delayed/scheduled tasks, crash recovery, result storage — but the ergonomics layer is still growing.

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)
• Automatic retries with quadratic backoff, configurable per task
• Crash recovery — in-flight tasks of a dead worker are reclaimed (XAUTOCLAIM)
• Results with TTL, plus task status (queued / running / complete / not_found)
• 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)

Installation

ArdiQ isn't on PyPI yet. Build it from source — you'll need Rust and uv:

$ git clone https://github.com/17tayyy/ardiq
$ cd ardiq
$ uv sync            # builds the Rust extension and installs dependencies

You also need a Redis server. For local development:

$ docker compose up -d

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.

Configuration

Ardiq(...) accepts:

Option	Default	Description
redis_url	redis://localhost:6379	Redis connection URL
queue_name	"default"	Logical queue (key namespace)
priorities	["default"]	Priority names, lowest-first
concurrency	16	Max tasks running at once
prefetch	concurrency * 2	Max tasks held in memory (drives backpressure)
idle_timeout_ms	60000	When an unrenewed in-flight task may be reclaimed
result_ttl_ms	300000	How long results live (0 drops, negative keeps forever)
burst	False	Exit once the queue drains
serializer / deserializer	msgpack	Wire codec; pass pickle.dumps/pickle.loads to send datetimes/objects

@app.task(...) accepts name, max_retries (default 3), backoff_ms, timeout (seconds), and priority. Use task.options(delay_ms=…, schedule_ms=…, priority=…, task_id=…).enqueue(...) for one-off overrides.

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.

License

MIT