litestar-rs

Distributed task queue on Redis Streams for Litestar — typed tasks, real dependency injection, retries, scheduling and cron, through Litestar's native plugin protocol.

📖 Documentation

Features

• Tasks are ordinary functions: arguments are serialised, dependencies injected
• Retries with backoff, a delivery ceiling, and a dead letter queue that keeps the payload, the traceback and the attempt history
• Delayed jobs and cron with no scheduler process, correct across DST
• Priority queues with a bounded starvation window, and shards for fairness
• Broker mode: consume streams somebody else writes, in the same worker
• Optional results, a deduplication gate, and trace context carried into the task
• Health endpoint served identically by the web process and the worker
• Eager mode, assert_enqueued and a real-worker fixture for your own tests

Installation

uv add litestar-rs

Redis 7 or newer. Standalone, Sentinel and Cluster are all covered by the test suite.

Quick start

from uuid import UUID

from litestar import Litestar, post
from litestar.di import Provide

from litestar_rs.plugin import QueueConfig, QueuePlugin, TaskRegistry

tasks = TaskRegistry()


@tasks.task
async def reindex(doc_id: UUID, session: AsyncSession) -> None:
    ...


@post("/documents")
async def create(doc_id: UUID) -> str:
    await reindex.enqueue(doc_id=doc_id)
    return "queued"


app = Litestar(
    route_handlers=[create],
    dependencies={"session": Provide(session)},
    plugins=[QueuePlugin(QueueConfig(registry=tasks, redis_url="redis://localhost"))],
)

doc_id travels in the payload and session is injected in the worker, from the application's own dependency graph. There is no context dictionary.

Workers

A worker is the same application, started differently:

litestar workers run --queue high --concurrency 20 --health-port 8081

Anything that can be settled at startup is — a task registered twice, an argument with no annotation, a dependency the application does not provide, a cycle, a provider only a request could satisfy — rather than on the first job in production.

Delivery guarantee

At-least-once. A worker that completed its side effect and died before XACK will be reclaimed and the work repeated. No amount of protocol work removes that; only an idempotent handler does. A dedup key is provided for the cases where that is not naturally true.

Documentation

Tasks, scheduling, retries, priorities, broker mode, results, testing and operations are covered in the full documentation. The rules the library is built to are in Design and invariants. Runnable examples are included.

Development

make install   # dependencies and git hooks
make check     # lint, types, import contracts, unit tests
make test-int  # integration suite, needs Docker

License

MIT