Metadata-Version: 2.4
Name: taskiq-redis-streams
Version: 0.2.0
Summary: A Redis Streams broker for Taskiq.
Keywords: async,redis,redis-streams,taskiq,tasks
Author: vvanglro
Author-email: vvanglro <vvanglro@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Distributed Computing
Requires-Dist: redis>=5,<9
Requires-Dist: taskiq>=0.12,<1
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/vvanglro/taskiq-redis-streams
Project-URL: Repository, https://github.com/vvanglro/taskiq-redis-streams
Project-URL: Issues, https://github.com/vvanglro/taskiq-redis-streams/issues
Description-Content-Type: text/markdown

# taskiq-redis-streams

[中文文档](README.zh-CN.md)

An independent, single-node Redis Streams broker for
[Taskiq](https://taskiq-python.github.io/). It uses bounded local prefetch and
consumer-heartbeat pending-entry recovery.

Redis Streams delivery is at least once. A task can execute more than once
after a worker crash or failed acknowledgement, so handlers must be idempotent.

## Installation

```bash
uv add taskiq-redis-streams
```

## Usage

```python
from taskiq_redis_streams import RedisAsyncResultBackend, RedisStreamsBroker

redis_url = "redis://localhost:6379/0"

result_backend = RedisAsyncResultBackend(
    redis_url,
    result_ex_time=3600,
    prefix_str="my-service:result",
)

broker = RedisStreamsBroker(
    redis_url,
    queue_name="default",
    namespace="my-service",
).with_result_backend(
    result_backend,
)


@broker.task
async def process_order(order_id: str) -> None:
    # Make this operation idempotent.
    ...
```

```bash
taskiq worker my_app:broker
```

The consumer group always starts at `0`, so tasks published before the first
worker starts are consumed.

## Features

- One namespaced Redis Stream and consumer group per Taskiq queue.
- At-least-once delivery through Redis Streams consumer groups; task handlers
  must be idempotent.
- Bounded broker-local prefetch: `max_pending` caps fetched-but-unacknowledged
  entries per listener.
- Consumer-heartbeat recovery: live workers can run arbitrarily long tasks
  without recovery being tied to a task execution timeout.
- Atomic orphan reclaim: Redis verifies that the PEL owner is unchanged and
  its heartbeat is absent before `XCLAIM` transfers an entry.
- A dedicated heartbeat thread and Redis connection to keep liveness renewal
  independent from ordinary event-loop stalls and blocking `XREADGROUP` calls.
- Fast listener-close handoff: entries still buffered inside the broker move to
  an internal `abandoned` consumer and are reclaimable immediately.
- Retryable Redis listener errors use exponential backoff; Taskiq cancellation
  is propagated so shutdown handoff still runs.
- Redis result backend with optional TTLs, progress storage, one-shot reads, and
  configurable key prefixes.

## Delivery and Recovery Flow

```mermaid
flowchart TD
    producer[Taskiq producer] -->|XADD data| stream[(Redis Stream)]
    startup[Worker startup] --> group[XGROUP CREATE consumer group]
    group --> listener[Broker listen loop]
    listener --> heartbeat[Background heartbeat thread refreshes TTL]
    heartbeat --> capacity{max_pending slot available?}
    capacity -->|no| wait_ack[Wait for a successful ACK]
    wait_ack --> capacity
    capacity -->|yes| reclaim_due{Reclaim scan due?}
    reclaim_due -->|yes| pending[XPENDING RANGE]
    pending --> lease{Owner heartbeat exists?}
    lease -->|no| claim[Lua: verify PEL owner then XCLAIM]
    lease -->|yes| read
    reclaim_due -->|no| read
    stream --> read[XREADGROUP new entries]
    claim --> buffer[Broker-local buffer]
    read --> buffer
    buffer --> deliver[Yield AckableMessage to Taskiq]
    deliver --> execute[Execute task]
    execute --> ack[XACK when Taskiq acknowledges]
    ack --> capacity

    listener_close[Listener cancellation or close] --> buffered{Still in local buffer?}
    buffered -->|yes| abandoned[XCLAIM to abandoned consumer]
    buffered -->|already yielded| drain[Keep heartbeat until broker shutdown]
    worker_loss[Crash or forced stop] --> expired[Heartbeat TTL expires]
    expired --> pending
    abandoned --> pending
```

## Behavior

One broker instance serves one Taskiq queue and uses these namespaced keys:

```text
<namespace>:stream:<queue_name>
<namespace>:workers:<queue_name>
<namespace>:heartbeat:<queue_name>:<consumer_name>
```

Use a distinct `namespace` for applications that share Redis. The default is
`taskiq`.

Each broker instance generates its own Redis consumer name. Consumer names are
not configurable, so a restarted worker receives a new identity. Active worker
consumers renew a Redis TTL heartbeat and pending entries owned by a consumer
whose heartbeat has expired are eligible for recovery.

`max_pending` caps entries fetched by a listener but not successfully
acknowledged; it defaults to `10`. Once the cap is reached, the listener leaves
new work for other consumers. Set `max_pending=1` to reserve one task at a time,
or `max_pending=None` to disable the local cap. Redis reads use an internal
batch size of at most `10`, always bounded by the remaining `max_pending`
capacity. Empty reads use an internal `3000` ms Redis long-poll timeout.

When `maxlen` is set, producers use Redis's approximate `XADD MAXLEN ~`
trimming to control Stream history. Choose it conservatively: trimming can
remove entries that have not yet been processed.

## Result Backend

`RedisAsyncResultBackend` is the single-node Redis result backend compatible
with Taskiq's `with_result_backend(...)` API. It stores serialized results and
progress under `<prefix_str>:<task_id>` and `<prefix_str>:<task_id>__progress`.
Without `prefix_str`, the task ID itself is the Redis key for compatibility with
`taskiq-redis`; set a distinct prefix when Redis is shared with other services.

Set exactly one of `result_ex_time` (seconds) or `result_px_time`
(milliseconds) to expire both results and progress. Leaving both unset keeps
them indefinitely. `keep_results=False` consumes a result atomically on its
first `get_result()` call. Use a TTL for long-running deployments to prevent
unbounded Redis storage.

`max_connection_pool_size` applies to task delivery commands. The broker keeps
one separate Redis connection in a heartbeat thread, so a blocking read or an
ordinary event-loop stall does not starve consumer lease renewal.

The broker scans the consumer group's PEL periodically. It renews each active
consumer's heartbeat every `consumer_heartbeat_interval` milliseconds (default
`10000`). A heartbeat remains live for `consumer_heartbeat_ttl` milliseconds
(default `60000`) after its most recent successful refresh. PEL scans run every
`reclaim_interval` milliseconds (default `10000`). Once that lease expires, the
next PEL scan claims entries from that consumer. `XCLAIM` is
guarded by an atomic Redis heartbeat check, so concurrent workers cannot both
recover the same entry. Configure a TTL longer than the renewal interval to
allow for normal scheduling and Redis latency. Reclaimed entries are handled
before new entries when a scan is due.

The heartbeat represents worker-process liveness, not task completion. A host
pause or network partition can still let the lease expire while a task resumes
later, so task handlers must remain idempotent.

Taskiq's `timeout` label still controls task execution time, but it does not
control Redis Streams recovery. A live worker can therefore run long tasks
without their PEL entries being reclaimed solely because of task duration.

On listener close, only messages still in the broker-local buffer are handed to
an internal `abandoned` consumer and become reclaimable immediately. Already
yielded messages may be executing. Their worker heartbeat remains active until
broker shutdown, allowing Taskiq to drain them without a duplicate reclaim.

Retryable Redis errors while listening are retried with exponential backoff from
100 ms to 5 seconds. Cancellation from Taskiq is never retried; it propagates
into the listener so the normal listener-close handoff can run.

Redis Cluster, Sentinel, delayed tasks, and a stream-level dead-letter queue
are not part of the first release.

## Development

Tests clear a dedicated Redis database before and after every test:

```bash
TEST_REDIS_URL=redis://127.0.0.1:7000/14 uv run pytest -q
```

Do not point `TEST_REDIS_URL` at a database containing application data.

```bash
uv sync --all-groups
uv run ruff check .
uv run mypy
uv run pytest -q
```

## Inspiration

The Taskiq integration follows ecosystem patterns from
[taskiq-redis](https://github.com/taskiq-python/taskiq-redis). Recovery, local
prefetch, and shutdown handoff are informed by
[dramatiq-redis-streams](https://github.com/sylvinus/dramatiq-redis-streams).

This project is independently maintained and is not affiliated with either
upstream project.
