taskiq-deduplication/docs/usage.md

Usage

Setup

Register RedisDeduplicationMiddleware on your broker before the application starts:

from taskiq_redis import ListQueueBroker
from taskiq_deduplication import RedisDeduplicationMiddleware

broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
    RedisDeduplicationMiddleware(redis_url="redis://localhost:6379"),
)

Middleware options

Parameter	Type	Default	Description
redis_url	str | RedisDsn	—	Redis connection URL passed to Redis.from_url. Accepts a plain string or a pydantic RedisDsn.
default_deduplication	bool	True	Whether deduplication is enabled for all tasks by default. Set False to opt-in per task instead of opting out.
default_ttl	int	300	Default lock TTL in seconds. Overridden per task with the deduplication_ttl label.
key_prefix	str	"taskiq:deduplication"	Prefix for all Redis lock keys.
startup_retries	int	3	Number of connection attempts during broker startup.
startup_retry_delay	float	1.0	Base delay in seconds between retries (exponential backoff: delay × 2^n).
heartbeat	bool	True	Whether to periodically re-extend the lock TTL while the task runs (see Long-running tasks).
heartbeat_interval	float | None	None	Seconds between heartbeat refreshes. When None, defaults to a third of the task's TTL (1s floor).
fail_open	bool	False	Whether a Redis error while acquiring the lock lets the task through instead of aborting the send (see Fail-open).

broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
    RedisDeduplicationMiddleware(
        redis_url="redis://localhost:6379",
        default_deduplication=True,
        default_ttl=60,
        key_prefix="myapp:dedup",
        startup_retries=5,
        startup_retry_delay=0.5,
    ),
)

Long-running tasks and the heartbeat

The lock is created with a TTL so a crashed worker cannot leak it forever. Without any refresh, a task that runs longer than its TTL would let the lock expire mid-execution, allowing a duplicate to be dispatched.

To prevent this, the middleware starts a background heartbeat in pre_execute that re-extends the lock TTL while the task runs (atomically, only if the lock is still owned by the running task). It is cancelled when the task completes or fails. This means you do not need to size default_ttl to your slowest task — the TTL only needs to outlive a single heartbeat interval; it acts purely as a safety net for worker crashes.

RedisDeduplicationMiddleware(
    redis_url="redis://localhost:6379",
    default_ttl=60,        # safety-net TTL; refreshed every ~20s while running
    heartbeat_interval=20, # optional; defaults to default_ttl / 3
)

If you disable the heartbeat (heartbeat=False), the invariant TTL must exceed the slowest task applies: set default_ttl (or the per-task deduplication_ttl label) above your worst-case task duration, or duplicates may slip through.

Startup resilience

On startup the middleware verifies the Redis connection with a PING. If Redis is temporarily unavailable, it retries with exponential backoff. After all attempts are exhausted a ConnectionError is raised and the broker fails to start.

Adjust startup_retries and startup_retry_delay to suit your deployment:

RedisDeduplicationMiddleware(
    redis_url="redis://localhost:6379",
    startup_retries=5,
    startup_retry_delay=2.0,
)

Fail-open

By default a Redis error while acquiring the lock aborts the send, so an unreachable Redis blocks task dispatch entirely. Set fail_open=True to trade deduplication for availability: the error is logged and the task is dispatched without a lock.

RedisDeduplicationMiddleware(
    redis_url="redis://localhost:6379",
    fail_open=True,
)

This applies to Redis errors only. A duplicate that is successfully detected still raises DuplicateTaskError, and while Redis is down duplicates can get through, so enable it only for tasks that tolerate running twice.

Redis errors after the task has been queued are always logged and swallowed, regardless of fail_open: failing to extend the lock after the send, to refresh it from the heartbeat, or to release it once the task ends never raises. Raising there would lose the result of a task that already ran; the lock expires on its TTL instead.

How it works

When a task is dispatched, the middleware acquires a Redis lock keyed on the task's fingerprint. Any subsequent dispatch with the same fingerprint raises DuplicateTaskError while the lock is held. The lock is released automatically when the task completes or fails.

The lock is acquired in two phases. It is first taken with a short grace TTL of 10 seconds, then extended to its full TTL once the broker has accepted the message. taskiq fires no middleware hook when the send itself fails, so this bounds the damage: if the broker is unreachable, the lock of a task that was never queued expires within seconds instead of blocking its fingerprint for the full TTL.

Handling duplicates

When a duplicate is detected, the middleware logs a warning and raises DuplicateTaskError, which prevents the task from reaching the broker. Catch it at the call site if you need to handle it explicitly:

from taskiq_deduplication import DuplicateTaskError

try:
    await my_task.kiq(user_id=42)
except DuplicateTaskError:
    pass  # task is already queued or running

DuplicateTaskError carries structured attributes describing the collision:

try:
    await my_task.kiq(user_id=42)
except DuplicateTaskError as err:
    logger.info(
        "Skipped %s; already held by %s (key=%s)",
        err.task_name,
        err.holder_task_id,
        err.key,
    )

• task_name — name of the task that was rejected.
• key — Redis lock key whose owner caused the rejection.
• holder_task_id — task_id of the task currently holding the lock, or None if it could not be retrieved.

Waiting for the winning task

holder_task_id is the task_id of the task that won the lock, so a rejected caller can build a handle to it and await its result instead of re-kicking:

from taskiq import AsyncTaskiqTask
from taskiq_deduplication import DuplicateTaskError

try:
    handle = await my_task.kiq(user_id=42)
except DuplicateTaskError as err:
    if err.holder_task_id is None:
        raise  # the lock was released in the meantime; retry the kiq() instead
    handle = AsyncTaskiqTask(err.holder_task_id, broker.result_backend)

result = await handle.wait_result()  # resolves when the winner finishes

Both callers now observe the same single execution, which is what you usually want from deduplication in a request handler: the second request waits for the first one's answer rather than being told to go away.

Three caveats:

• The result backend must be shared and persistent. InmemoryResultBackend only works within a single process; across processes the loser cannot see the winner's result.
• The winner's result must not have expired. If your backend sets a result TTL, a loser that waits longer than that gets nothing back.
• holder_task_id can be None, when the lock is released between the failed SET NX and the follow-up GET. Fall back to re-kicking, as above: the lock is free again, so the retry acquires it.

Per-task label overrides

Labels can be set at the task level (applied to every call) or at call time.

Task-level (decorator)

@broker.task(deduplication_ttl=60)
async def my_task(user_id: int) -> None:
    ...

Call-level (kicker)

await my_task.kicker().with_labels(deduplication_ttl=60).kiq(user_id=42)

Available labels

Label	Type	Description
deduplication	bool	Set False to opt out of deduplication entirely for this task.
deduplication_ttl	int	Lock TTL in seconds. Overrides the middleware default_ttl.
deduplication_key	str	Explicit lock key. Skips fingerprint computation entirely.
deduplication_key_fields	list[str]	Subset of kwargs to include in the fingerprint. Positional arguments are excluded. Ignored if deduplication_key is set.

Fingerprint and key customisation

By default the lock key is a SHA-256 fingerprint of the task name, its positional arguments and all kwargs.

!!! warning "Positional and keyword calls fingerprint differently"

taskiq serialises arguments as they were passed, without binding them to the
task signature. `my_task.kiq(42)` and `my_task.kiq(user_id=42)` are therefore
*not* recognised as duplicates of each other. Call a deduplicated task
consistently, preferably always with keyword arguments.

Explicit key

Use deduplication_key when you want full control over the lock key, regardless of the kwargs:

@broker.task(deduplication_key="send-welcome-email")
async def send_welcome_email(user_id: int, locale: str) -> None:
    ...

All calls to this task share a single lock, no matter what arguments are passed.

Partial key (key fields)

Use deduplication_key_fields to deduplicate only on a subset of kwargs. Here, two calls with the same user_id but different locale are treated as duplicates:

@broker.task(deduplication_key_fields=["user_id"])
async def send_welcome_email(user_id: int, locale: str) -> None:
    ...

If a listed field is absent from a task's kwargs, it is dropped from the fingerprint and a warning is logged, since this can make genuinely different calls collide on the same lock.

Positional arguments are excluded from the fingerprint entirely when this label is set: you asked to deduplicate on named fields, so pass them as keyword arguments. A warning is logged if the task is called with positional arguments anyway.

Opting out per task

@broker.task(deduplication=False)
async def always_run(payload: str) -> None:
    ...