Version 1.2.0

* ⬆ bump pytest from 9.0.3 to 9.1.1

Bumps [pytest](https://github.com/pytest-dev/pytest) from 9.0.3 to 9.1.1.
- [Release notes](https://github.com/pytest-dev/pytest/releases)
- [Changelog](https://github.com/pytest-dev/pytest/blob/main/CHANGELOG.rst)
- [Commits](https://github.com/pytest-dev/pytest/compare/9.0.3...9.1.1)

---
updated-dependencies:
- dependency-name: pytest
  dependency-version: 9.1.1
  dependency-type: direct:development
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

* fix: fake_redis pytest fixture

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: d3vyce <nicolas.sudres@proton.me>

.github/workflows/build-release.yml

@@ -11,7 +11,7 @@ jobs:
     permissions:
       id-token: write
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
 
       - name: Install uv
         uses: astral-sh/setup-uv@v7

.github/workflows/ci.yml

@@ -18,7 +18,7 @@ jobs:
     name: Lint (Ruff)
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
 
       - name: Install uv
         uses: astral-sh/setup-uv@v7
@@ -39,7 +39,7 @@ jobs:
     name: Type Check (ty)
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
 
       - name: Install uv
         uses: astral-sh/setup-uv@v7
@@ -62,7 +62,7 @@ jobs:
         python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
 
       - name: Install uv
         uses: astral-sh/setup-uv@v7
@@ -73,12 +73,12 @@ jobs:
       - name: Install dependencies
         run: uv sync --group dev
 
-      - name: Run tests with coverage
-        run: uv run pytest --cov --cov-report=xml --cov-report=term-missing  --junitxml=junit.xml -o junit_family=legacy
+      - name: Run unit tests with coverage
+        run: uv run pytest -m "not integration" --cov --cov-report=xml --cov-report=term-missing --junitxml=junit.xml -o junit_family=legacy
 
       - name: Upload coverage to Codecov
         if: matrix.python-version == '3.14'
-        uses: codecov/codecov-action@v6
+        uses: codecov/codecov-action@v7
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
           report_type: coverage
@@ -87,7 +87,43 @@ jobs:
 
       - name: Upload test results to Codecov
         if: matrix.python-version == '3.14'
-        uses: codecov/codecov-action@v6
+        uses: codecov/codecov-action@v7
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
           report_type: test_results
+
+  test-integration:
+    name: Integration (${{ matrix.backend.name }})
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        backend:
+          - { name: redis, image: "redis:7" }
+          - { name: valkey, image: "valkey/valkey:8" }
+
+    services:
+      store:
+        image: ${{ matrix.backend.image }}
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 3s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+
+      - name: Set up Python
+        run: uv python install 3.13
+
+      - name: Install dependencies
+        run: uv sync --group dev
+
+      - name: Run integration tests
+        run: uv run pytest -m integration -v

.github/workflows/docs.yml

@@ -16,9 +16,9 @@ jobs:
       url: ${{ steps.deployment.outputs.page_url }}
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/configure-pages@v5
+      - uses: actions/configure-pages@v6
 
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
 
       - name: Install uv
         uses: astral-sh/setup-uv@v7
@@ -30,9 +30,9 @@ jobs:
 
       - run: uv run zensical build --clean
 
-      - uses: actions/upload-pages-artifact@v4
+      - uses: actions/upload-pages-artifact@v5
         with:
           path: site
 
-      - uses: actions/deploy-pages@v4
+      - uses: actions/deploy-pages@v5
         id: deployment

README.md

@@ -21,7 +21,7 @@ Redis-backed deduplication middleware for Taskiq that prevents duplicate tasks f
 ## Installation
 
 ```bash
-uv add taskiq-deduplication
+uv add "taskiq-deduplication"
 ```
 
 ## Quick Start
@@ -34,9 +34,10 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
     RedisDeduplicationMiddleware(redis_url="redis://localhost:6379"),
 )
 
+
 @broker.task
-async def send_report(user_id: int) -> None:
-    ...
+async def send_report(user_id: int) -> None: ...
+
 
 # First dispatch acquires the lock — succeeds.
 await send_report.kiq(user_id=42)
@@ -50,12 +51,16 @@ except DuplicateTaskError:
 
 ## Features
 
-- **Sender-side deduplication** — rejects duplicate tasks at dispatch time via a Redis queue lock, before they reach the broker.
-- **Worker-side detection** — logs concurrent duplicate executions without raising, keeping `SmartRetryMiddleware` safe from retry storms.
+- **Sender-side deduplication** — rejects duplicate tasks at dispatch time via a Redis lock, before they reach the broker.
+- **Handle to the winning task** — a rejected caller gets the winner's `task_id` on the error, so it can await the winner's result instead of re-kicking.
+- **Atomic lock release** — lock is released on completion or error via a Lua check-and-delete; only the owning task can release its lock.
 - **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
+- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
 - **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
-- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
 - **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
+- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
+- **Fail-open mode** — opt in with `fail_open` to keep dispatching tasks when Redis is unreachable at runtime, trading deduplication for availability.
 
 ## License
 

docs/index.md

@@ -34,9 +34,10 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
     RedisDeduplicationMiddleware(redis_url="redis://localhost:6379"),
 )
 
+
 @broker.task
-async def send_report(user_id: int) -> None:
-    ...
+async def send_report(user_id: int) -> None: ...
+
 
 # First dispatch acquires the lock — succeeds.
 await send_report.kiq(user_id=42)
@@ -50,12 +51,16 @@ except DuplicateTaskError:
 
 ## Features
 
-- **Sender-side deduplication** — rejects duplicate tasks at dispatch time via a Redis queue lock, before they reach the broker.
-- **Worker-side detection** — logs concurrent duplicate executions without raising, keeping `SmartRetryMiddleware` safe from retry storms.
+- **Sender-side deduplication** — rejects duplicate tasks at dispatch time via a Redis lock, before they reach the broker.
+- **Handle to the winning task** — a rejected caller gets the winner's `task_id` on the error, so it can await the winner's result instead of re-kicking.
+- **Atomic lock release** — lock is released on completion or error via a Lua check-and-delete; only the owning task can release its lock.
 - **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
+- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
 - **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
-- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
 - **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
+- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
+- **Fail-open mode** — opt in with `fail_open` to keep dispatching tasks when Redis is unreachable at runtime, trading deduplication for availability.
 
 ## License
 

docs/overrides/main.html

@@ -0,0 +1,7 @@
+{% extends "base.html" %} {% block extrahead %}
+<script
+    defer
+    src="https://analytics.d3vyce.fr/script.js"
+    data-website-id="07b1421a-a466-4816-a55b-ae76a101cb4b"
+></script>
+{{ super() }} {% endblock %}

docs/reference.md

@@ -1,13 +1,14 @@
 # API Reference
 
-## Middleware
+You can import them directly from `taskiq_deduplication`:
 
-::: taskiq_deduplication.RedisDeduplicationMiddleware
-    options:
-      show_source: false
+```python
+from taskiq_deduplication import (
+    DuplicateTaskError,
+    RedisDeduplicationMiddleware,
+)
+```
 
-## Exceptions
+## ::: taskiq_deduplication.RedisDeduplicationMiddleware
 
-::: taskiq_deduplication.DuplicateTaskError
-    options:
-      show_source: false
+## ::: taskiq_deduplication.DuplicateTaskError

docs/usage.md

@@ -17,10 +17,15 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
 
 | Parameter | Type | Default | Description |
 |---|---|---|---|
-| `redis_url` | `str` | — | Redis connection URL passed to `Redis.from_url`. |
+| `redis_url` | `str \| RedisDsn` | — | Redis connection URL passed to `Redis.from_url`. Accepts a plain string or a pydantic [`RedisDsn`](https://docs.pydantic.dev/latest/api/networks/#pydantic.networks.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](#long-running-tasks-and-the-heartbeat)). |
+| `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](#fail-open)). |
 
 ```python
 broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
@@ -29,10 +34,77 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
         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.
+
+```python
+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:
+
+```python
+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.
+
+```python
+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
@@ -40,6 +112,12 @@ 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
@@ -55,6 +133,59 @@ except DuplicateTaskError:
     pass  # task is already queued or running
 ```
 
+`DuplicateTaskError` carries structured attributes describing the collision:
+
+```python
+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:
+
+```python
+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.
@@ -63,8 +194,7 @@ Labels can be set at the task level (applied to every call) or at call time.
 
 ```python
 @broker.task(deduplication_ttl=60)
-async def my_task(user_id: int) -> None:
-    ...
+async def my_task(user_id: int) -> None: ...
 ```
 
 ### Call-level (kicker)
@@ -80,11 +210,19 @@ await my_task.kicker().with_labels(deduplication_ttl=60).kiq(user_id=42)
 | `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. Ignored if `deduplication_key` is set. |
+| `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 and all kwargs.
+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
 
@@ -93,8 +231,7 @@ the kwargs:
 
 ```python
 @broker.task(deduplication_key="send-welcome-email")
-async def send_welcome_email(user_id: int, locale: str) -> None:
-    ...
+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.
@@ -107,14 +244,20 @@ duplicates:
 
 ```python
 @broker.task(deduplication_key_fields=["user_id"])
-async def send_welcome_email(user_id: int, locale: str) -> None:
-    ...
+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
 
 ```python
 @broker.task(deduplication=False)
-async def always_run(payload: str) -> None:
-    ...
+async def always_run(payload: str) -> None: ...
 ```

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "taskiq-deduplication"
-version = "1.0.0"
-description = "Production-ready utilities for FastAPI applications"
+version = "1.2.0"
+description = "Redis-backed deduplication middleware for Taskiq"
 readme = "README.md"
 license = "MIT"
 license-files = ["LICENSE"]
@@ -9,7 +9,7 @@ requires-python = ">=3.10"
 authors = [
     { name = "d3vyce", email = "contact@d3vyce.fr" }
 ]
-keywords = ["fastapi", "sqlalchemy", "postgresql"]
+keywords = ["taskiq", "redis", "deduplication", "middleware", "task-queue"]
 classifiers = [
     "Development Status :: 5 - Production/Stable",
     "Framework :: AsyncIO",
@@ -27,6 +27,7 @@ classifiers = [
     "Typing :: Typed",
 ]
 dependencies = [
+    "pydantic>=2.0.0",
     "redis>=7.0.0",
     "taskiq>=0.12.0",
 ]
@@ -44,11 +45,11 @@ dev = [
     "taskiq_deduplication",
     "prek>=0.3.8",
     "ruff>=0.1.0",
-    "ty>=0.0.1a0",
+    "ty>=0.0.34",
 ]
 tests = [
     "coverage>=7.0.0",
-    "fakeredis[lua]>=2.0.0",
+    "fakeredis[lua]>=2.35.1",
     "pytest-anyio>=0.0.0",
     "pytest-cov>=4.0.0",
     "pytest>=8.0.0",
@@ -59,14 +60,21 @@ docs = [
 ]
 
 [build-system]
-requires = ["uv_build>=0.10,<0.12.0"]
+requires = ["uv_build>=0.11.8,<0.13.0"]
 build-backend = "uv_build"
 
+[tool.ruff.lint]
+ignore = ["BLE001"]
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+anyio_mode = "auto"
 filterwarnings = [
     "ignore::DeprecationWarning",
 ]
+markers = [
+    "integration: requires a live Redis at localhost:6379",
+]
 
 [tool.coverage.run]
 source = ["src/taskiq_deduplication"]

src/taskiq_deduplication/__init__.py

@@ -1,8 +1,8 @@
-"""FastAPI utilities package."""
+"""Redis-backed deduplication middleware for Taskiq."""
 
 from .middleware import DuplicateTaskError, RedisDeduplicationMiddleware
 
-__version__ = "1.0.0"
+__version__ = "1.2.0"
 
 __all__ = [
     "DuplicateTaskError",

src/taskiq_deduplication/middleware.py

@@ -1,13 +1,23 @@
+import asyncio
 import hashlib
 import json
 import logging
 from typing import Any
 
+from pydantic import RedisDsn
 from redis.asyncio import Redis
 from taskiq import TaskiqMessage, TaskiqResult
 from taskiq.abc.middleware import TaskiqMiddleware
 
-from .utils import check_and_delete
+from .utils import (
+    REFRESH_LUA_SCRIPT,
+    RELEASE_LUA_SCRIPT,
+    check_and_delete,
+    check_and_refresh,
+    parse_bool_label,
+    parse_int_label,
+    parse_list_label,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -15,10 +25,35 @@ DEDUP_LABEL = "deduplication"
 DEDUP_TTL_LABEL = "deduplication_ttl"
 DEDUP_KEY_FIELDS_LABEL = "deduplication_key_fields"
 DEDUP_EXPLICIT_KEY_LABEL = "deduplication_key"
+SEND_GRACE_TTL = 10
+
+_CACHED_KEY_LABEL = "__taskiq_dedup_cached_key"
 
 
 class DuplicateTaskError(Exception):
-    """Raised when a task with identical name and kwargs is already queued or running."""
+    """Raised when a task with identical name and kwargs is already queued or running.
+
+    Attributes:
+        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 (e.g. the lock was released
+            between the failed acquisition and the lookup).
+    """
+
+    def __init__(
+        self,
+        task_name: str,
+        key: str,
+        holder_task_id: str | None = None,
+    ) -> None:
+        self.task_name = task_name
+        self.key = key
+        self.holder_task_id = holder_task_id
+        super().__init__(
+            f"Task {task_name!r} with the same arguments is already queued or "
+            f"running (key={key!r}, holder_task_id={holder_task_id!r})."
+        )
 
 
 class RedisDeduplicationMiddleware(TaskiqMiddleware):
@@ -30,97 +65,315 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
     on completion or error.
 
     Attributes:
-        redis_url: Redis connection URL passed to ``Redis.from_url``.
+        redis_url: Redis connection URL (``str`` or ``RedisDsn``) passed to
+            ``Redis.from_url``.
         default_deduplication: Whether deduplication is enabled by default.
         default_ttl: Default lock TTL in seconds.
         key_prefix: Prefix for all Redis lock keys.
+        heartbeat: Whether to periodically re-extend the lock TTL during task
+            execution so long-running tasks keep their lock.
+        heartbeat_interval: Seconds between heartbeat refreshes. When ``None`` it
+            defaults to a third of the task's TTL (with a 1s floor).
+        fail_open: Whether a Redis error while acquiring the lock lets the task
+            through instead of aborting the send. Detected duplicates still raise
+            ``DuplicateTaskError``.
     """
 
     def __init__(
         self,
-        redis_url: str,
+        redis_url: str | RedisDsn,
         default_deduplication: bool = True,
         default_ttl: int = 300,
         key_prefix: str = "taskiq:deduplication",
+        startup_retries: int = 3,
+        startup_retry_delay: float = 1.0,
+        heartbeat: bool = True,
+        heartbeat_interval: float | None = None,
+        fail_open: bool = False,
     ) -> None:
         self.redis_url = redis_url
         self.default_deduplication = default_deduplication
         self.default_ttl = default_ttl
         self.key_prefix = key_prefix
+        self.startup_retries = startup_retries
+        self.startup_retry_delay = startup_retry_delay
+        self.heartbeat = heartbeat
+        self.heartbeat_interval = heartbeat_interval
+        self.fail_open = fail_open
         self._redis: Redis | None = None
+        self._release_script: Any = None
+        self._refresh_script: Any = None
+        self._heartbeats: dict[str, asyncio.Task[None]] = {}
 
     async def startup(self) -> None:
-        self._redis = Redis.from_url(self.redis_url)
+        last_error: BaseException | None = None
+        for attempt in range(self.startup_retries):
+            client = Redis.from_url(str(self.redis_url))
+            try:
+                await client.ping()
+                self._redis = client
+                self._release_script = self._redis.register_script(RELEASE_LUA_SCRIPT)
+                self._refresh_script = self._redis.register_script(REFRESH_LUA_SCRIPT)
+                return
+            except Exception as exc:
+                await client.aclose()
+                last_error = exc
+                if attempt < self.startup_retries - 1:
+                    delay = self.startup_retry_delay * (2**attempt)
+                    logger.warning(
+                        "Failed to connect to Redis (attempt %d/%d): %s. "
+                        "Retrying in %.1fs...",
+                        attempt + 1,
+                        self.startup_retries,
+                        exc,
+                        delay,
+                    )
+                    await asyncio.sleep(delay)
+        logger.error(
+            "Failed to connect to Redis after %d attempts.", self.startup_retries
+        )
+        raise ConnectionError(
+            f"Could not connect to Redis after {self.startup_retries} attempts"
+        ) from last_error
 
     async def shutdown(self) -> None:
+        for task_id in list(self._heartbeats):
+            await self._cancel_heartbeat(task_id)
         if self._redis is not None:
             await self._redis.aclose()
 
-    def _build_deduplication_key(self, message: TaskiqMessage) -> str:
+    def _build_deduplication_key(self, message: TaskiqMessage) -> str | None:
         explicit_key: str | None = message.labels.get(DEDUP_EXPLICIT_KEY_LABEL)
         if explicit_key is not None:
             return f"{self.key_prefix}:{explicit_key}"
 
-        key_fields: list[str] | None = message.labels.get(DEDUP_KEY_FIELDS_LABEL)
-        kwargs = (
-            {k: v for k, v in message.kwargs.items() if k in key_fields}
-            if key_fields is not None
-            else message.kwargs
-        )
-        payload = json.dumps(
-            {"task": message.task_name, "kwargs": kwargs},
-            sort_keys=True,
+        key_fields = parse_list_label(
+            message.labels.get(DEDUP_KEY_FIELDS_LABEL), DEDUP_KEY_FIELDS_LABEL
         )
+        if key_fields is not None:
+            missing = [field for field in key_fields if field not in message.kwargs]
+            if missing:
+                logger.warning(
+                    "Task %s requested deduplication_key_fields %r but they are "
+                    "absent from kwargs; they are dropped from the fingerprint, which "
+                    "may cause distinct calls to collide.",
+                    message.task_name,
+                    missing,
+                )
+            if message.args:
+                logger.warning(
+                    "Task %s was called with positional arguments but uses "
+                    "deduplication_key_fields; positional arguments are excluded "
+                    "from the fingerprint.",
+                    message.task_name,
+                )
+            kwargs = {k: v for k, v in message.kwargs.items() if k in key_fields}
+            args: list[Any] = []
+        else:
+            kwargs = message.kwargs
+            args = message.args
+        try:
+            payload = json.dumps(
+                {"task": message.task_name, "args": args, "kwargs": kwargs},
+                sort_keys=True,
+            )
+        except TypeError:
+            return None
         fingerprint = hashlib.sha256(payload.encode()).hexdigest()[:16]
         return f"{self.key_prefix}:{fingerprint}"
 
     def _is_enabled(self, labels: dict[str, Any]) -> bool:
-        return bool(labels.get(DEDUP_LABEL, self.default_deduplication))
+        return parse_bool_label(
+            labels.get(DEDUP_LABEL), self.default_deduplication, DEDUP_LABEL
+        )
 
     def _get_ttl(self, labels: dict[str, Any]) -> int:
-        return int(labels.get(DEDUP_TTL_LABEL, self.default_ttl))
+        return parse_int_label(
+            labels.get(DEDUP_TTL_LABEL, self.default_ttl),
+            self.default_ttl,
+            DEDUP_TTL_LABEL,
+        )
 
     async def _release_if_owned(self, key: str, task_id: str) -> None:
-        assert self._redis is not None
-        released = await check_and_delete(self._redis, key, task_id)
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
+        if self._release_script is None:
+            self._release_script = self._redis.register_script(RELEASE_LUA_SCRIPT)
+        released = await check_and_delete(self._release_script, key, task_id)
         if released:
             logger.debug("Released lock %s", key)
         else:
             logger.debug("Skipped release of lock %s: not owned by this task", key)
 
+    async def _refresh_if_owned(self, key: str, task_id: str, ttl: int) -> bool:
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
+        if self._refresh_script is None:
+            self._refresh_script = self._redis.register_script(REFRESH_LUA_SCRIPT)
+        return await check_and_refresh(self._refresh_script, key, task_id, ttl)
+
+    @staticmethod
+    def _get_cached_key(message: TaskiqMessage) -> str | None:
+        return message.labels.get(_CACHED_KEY_LABEL)
+
+    @staticmethod
+    def _cache_key(message: TaskiqMessage, key: str | None) -> None:
+        message.labels[_CACHED_KEY_LABEL] = key
+
     async def pre_send(self, message: TaskiqMessage) -> TaskiqMessage:
         if not self._is_enabled(message.labels):
             return message
 
-        assert self._redis is not None
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
         key = self._build_deduplication_key(message)
+        self._cache_key(message, key)
+        if key is None:
+            logger.warning(
+                "Task %s has non-JSON-serializable kwargs; deduplication skipped."
+                " Use the deduplication_key label to deduplicate this task.",
+                message.task_name,
+            )
+            return message
         ttl = self._get_ttl(message.labels)
 
         logger.debug("Acquiring lock %s for task %s", key, message.task_name)
-        acquired = await self._redis.set(key, message.task_id, ex=ttl, nx=True)
-        if not acquired:
+        try:
+            acquired = await self._redis.set(
+                key, message.task_id, ex=min(ttl, SEND_GRACE_TTL), nx=True
+            )
+            holder_task_id = None if acquired else await self._redis.get(key)
+        except Exception as exc:
+            if not self.fail_open:
+                raise
             logger.warning(
-                "Duplicate task %s dropped (key=%s).",
+                "Redis is unavailable (%s); dispatching task %s without "
+                "deduplication (fail_open is enabled).",
+                exc,
+                message.task_name,
+            )
+            # No lock was taken: nothing downstream should release or refresh one.
+            self._cache_key(message, None)
+            return message
+
+        if not acquired:
+            if isinstance(holder_task_id, bytes):
+                holder_task_id = holder_task_id.decode()
+            logger.warning(
+                "Duplicate task %s dropped (key=%s, holder_task_id=%s).",
                 message.task_name,
                 key,
+                holder_task_id,
             )
             raise DuplicateTaskError(
-                f"Task {message.task_name!r} with the same arguments is already queued or running."
+                task_name=message.task_name,
+                key=key,
+                holder_task_id=holder_task_id,
             )
 
         logger.debug("Lock %s acquired for task %s", key, message.task_name)
         return message
 
+    async def post_send(self, message: TaskiqMessage) -> None:
+        # The cached key is set by pre_send() only when deduplication is enabled.
+        key = self._get_cached_key(message)
+        if key is None:
+            return
+        ttl = self._get_ttl(message.labels)
+        if ttl <= SEND_GRACE_TTL:
+            return
+        try:
+            # Returns False when a fast worker already ran and released the lock.
+            extended = await self._refresh_if_owned(key, message.task_id, ttl)
+        except Exception as exc:
+            # The task is already queued; never fail the send. The lock just keeps
+            # its grace TTL.
+            logger.warning("Failed to extend lock %s after send: %s", key, exc)
+            return
+        logger.debug(
+            "Lock %s %s to the full TTL (%ds) after send",
+            key,
+            "extended" if extended else "not extended",
+            ttl,
+        )
+
+    def _get_heartbeat_interval(self, ttl: int) -> float:
+        if self.heartbeat_interval is not None:
+            return self.heartbeat_interval
+        return max(ttl / 3, 1.0)
+
+    async def _heartbeat_loop(
+        self, key: str, task_id: str, ttl: int, interval: float
+    ) -> None:
+        try:
+            while True:
+                await asyncio.sleep(interval)
+                try:
+                    refreshed = await self._refresh_if_owned(key, task_id, ttl)
+                except Exception as exc:
+                    logger.warning("Failed to refresh lock %s: %s", key, exc)
+                    continue
+                if refreshed:
+                    logger.debug("Refreshed lock %s (ttl=%ds)", key, ttl)
+                else:
+                    logger.warning(
+                        "Lock %s no longer owned by task %s; stopping heartbeat.",
+                        key,
+                        task_id,
+                    )
+                    return
+        except asyncio.CancelledError:
+            pass
+
+    async def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
+        if not self.heartbeat:
+            return message
+        # The cached key is set by pre_send() only when deduplication is enabled.
+        key = self._get_cached_key(message)
+        if key is None:
+            return message
+        ttl = self._get_ttl(message.labels)
+        interval = self._get_heartbeat_interval(ttl)
+        self._heartbeats[message.task_id] = asyncio.create_task(
+            self._heartbeat_loop(key, message.task_id, ttl, interval)
+        )
+        return message
+
+    async def _cancel_heartbeat(self, task_id: str) -> None:
+        task = self._heartbeats.pop(task_id, None)
+        if task is None:
+            return
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    async def _release_lock(self, message: TaskiqMessage) -> None:
+        await self._cancel_heartbeat(message.task_id)
+        # The cached key is set by pre_send() only when deduplication is enabled.
+        key = self._get_cached_key(message)
+        if key is None:
+            return
+        try:
+            await self._release_if_owned(key, message.task_id)
+        except Exception as exc:
+            # The task already ran: raising here would lose its result in the
+            # receiver. The lock expires on its TTL instead.
+            logger.warning("Failed to release lock %s: %s", key, exc)
+
     async def post_execute(
         self,
         message: TaskiqMessage,
         result: TaskiqResult,
     ) -> None:
-        if not self._is_enabled(message.labels):
-            return
-        await self._release_if_owned(
-            self._build_deduplication_key(message), message.task_id
-        )
+        await self._release_lock(message)
 
     async def on_error(
         self,
@@ -128,8 +381,4 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         result: TaskiqResult,
         exception: BaseException,
     ) -> None:
-        if not self._is_enabled(message.labels):
-            return
-        await self._release_if_owned(
-            self._build_deduplication_key(message), message.task_id
-        )
+        await self._release_lock(message)

src/taskiq_deduplication/utils.py

@@ -1,18 +1,104 @@
-from collections.abc import Awaitable
-from typing import cast
+import ast
+import logging
+from typing import Any
 
-from redis.asyncio import Redis
+logger = logging.getLogger(__name__)
+
+RELEASE_LUA_SCRIPT = """
+if redis.call('get', KEYS[1]) == ARGV[1] then
+    return redis.call('del', KEYS[1])
+else
+    return 0
+end
+"""
+
+REFRESH_LUA_SCRIPT = """
+if redis.call('get', KEYS[1]) == ARGV[1] then
+    return redis.call('expire', KEYS[1], ARGV[2])
+else
+    return 0
+end
+"""
 
 
-async def check_and_delete(redis: Redis, key: str, owner: str) -> bool:
-    """Delete *key* only if its value equals *owner*. Returns True if deleted."""
-    release_script = """
-    if redis.call('get', KEYS[1]) == ARGV[1] then
-        return redis.call('del', KEYS[1])
-    else
-        return 0
-    end
+async def check_and_delete(script: Any, key: str, owner: str) -> bool:
+    """Delete *key* only if its value equals *owner*.
+
+    Args:
+        script: Pre-registered Lua script object (from ``Redis.register_script``).
+        key: Lock key to delete.
+        owner: Expected value of the key (task_id).
+
+    Returns:
+        True if the key was deleted, False otherwise.
     """
-
-    released = await cast(Awaitable[int], redis.eval(release_script, 1, key, owner))
+    released: int = await script(keys=[key], args=[owner])
     return bool(released)
+
+
+async def check_and_refresh(script: Any, key: str, owner: str, ttl: int) -> bool:
+    """Extend *key*'s TTL to *ttl* only if its value equals *owner*.
+
+    Args:
+        script: Pre-registered Lua script object (from ``Redis.register_script``).
+        key: Lock key to refresh.
+        owner: Expected value of the key (task_id).
+        ttl: New TTL in seconds.
+
+    Returns:
+        True if the TTL was extended, False if the key is missing or owned by
+        another task.
+    """
+    refreshed: int = await script(keys=[key], args=[owner, ttl])
+    return bool(refreshed)
+
+
+def parse_bool_label(value: Any, default: bool, label_name: str = "") -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        lower = value.lower()
+        if lower == "true":
+            return True
+        if lower == "false":
+            return False
+    if value is not None:
+        logger.warning(
+            "Invalid %r value %r (expected bool); falling back to default (%r).",
+            label_name,
+            value,
+            default,
+        )
+    return default
+
+
+def parse_list_label(value: Any, label_name: str = "") -> list[str] | None:
+    if isinstance(value, list):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = ast.literal_eval(value)
+            if isinstance(parsed, list):
+                return parsed
+        except (ValueError, SyntaxError):
+            pass
+    if value is not None:
+        logger.warning(
+            "Invalid %r value %r (expected list[str]); ignoring.",
+            label_name,
+            value,
+        )
+    return None
+
+
+def parse_int_label(value: Any, default: int, label_name: str = "") -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        logger.warning(
+            "Invalid %r value %r (expected int); falling back to default (%d).",
+            label_name,
+            value,
+            default,
+        )
+        return default

tests/conftest.py

@@ -1,5 +1,6 @@
-import pytest
 import fakeredis.aioredis
+import pytest
+from redis.asyncio import Redis
 from taskiq import TaskiqMessage, TaskiqResult
 
 
@@ -9,21 +10,36 @@ def anyio_backend():
 
 
 @pytest.fixture
-async def fake_redis():
-    client = fakeredis.aioredis.FakeRedis()
+def fake_redis():
+    return fakeredis.aioredis.FakeRedis()
+
+
+@pytest.fixture
+async def real_redis():
+    client = Redis.from_url("redis://localhost:6379/15")
+    try:
+        await client.ping()
+    except Exception:
+        await client.aclose()
+        pytest.skip("Redis not available at localhost:6379")
+        return
+    await client.flushdb()
     yield client
+    await client.flushdb()
     await client.aclose()
 
 
 @pytest.fixture
 def make_message():
-    def _make(task_name="my_task", task_id="task-1", labels=None, kwargs=None):
+    def _make(
+        task_name="my_task", task_id="task-1", labels=None, kwargs=None, args=None
+    ):
         return TaskiqMessage(
             task_id=task_id,
             task_name=task_name,
             labels=labels or {},
             labels_types={},
-            args=[],
+            args=args or [],
             kwargs=kwargs or {},
         )
 

tests/test_integration.py

@@ -0,0 +1,116 @@
+"""Integration tests against a live Redis instance (localhost:6379/15).
+
+These tests are skipped automatically when Redis is not reachable.
+In CI a Redis service is started before the test step.
+"""
+
+import pytest
+
+from taskiq_deduplication import DuplicateTaskError, RedisDeduplicationMiddleware
+from taskiq_deduplication.middleware import SEND_GRACE_TTL
+from taskiq_deduplication.utils import REFRESH_LUA_SCRIPT, RELEASE_LUA_SCRIPT
+
+
+@pytest.fixture
+def mw(real_redis):
+    middleware = RedisDeduplicationMiddleware(redis_url="redis://localhost:6379/15")
+    middleware._redis = real_redis
+    middleware._release_script = real_redis.register_script(RELEASE_LUA_SCRIPT)
+    middleware._refresh_script = real_redis.register_script(REFRESH_LUA_SCRIPT)
+    return middleware
+
+
+@pytest.mark.integration
+async def test_full_lifecycle(mw, real_redis, make_message, make_result):
+    msg = make_message()
+    await mw.pre_send(msg)
+    key = mw._build_deduplication_key(msg)
+    assert await real_redis.exists(key)
+
+    await mw.post_execute(msg, make_result())
+    assert not await real_redis.exists(key)
+
+    await mw.pre_send(make_message())
+
+
+@pytest.mark.integration
+async def test_duplicate_rejected(mw, make_message):
+    await mw.pre_send(make_message())
+    with pytest.raises(DuplicateTaskError):
+        await mw.pre_send(make_message())
+
+
+@pytest.mark.integration
+async def test_redispatch_after_on_error(mw, make_message, make_result):
+    msg = make_message()
+    await mw.pre_send(msg)
+    await mw.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+    await mw.pre_send(make_message())
+
+
+@pytest.mark.integration
+async def test_lua_only_owner_can_release(mw, real_redis, make_message):
+    owner_msg = make_message(task_id="owner")
+    key = mw._build_deduplication_key(owner_msg)
+    await real_redis.set(key, "owner", ex=60)
+
+    await mw._release_if_owned(key, "intruder")
+    assert await real_redis.exists(key)
+
+    await mw._release_if_owned(key, "owner")
+    assert not await real_redis.exists(key)
+
+
+@pytest.mark.integration
+async def test_ttl_is_applied(mw, real_redis, make_message):
+    msg = make_message()
+    await mw.pre_send(msg)
+    await mw.post_send(msg)
+    key = mw._build_deduplication_key(msg)
+    ttl = await real_redis.ttl(key)
+    assert SEND_GRACE_TTL < ttl <= mw.default_ttl
+
+
+@pytest.mark.integration
+async def test_heartbeat_keeps_long_running_lock_alive(
+    mw, real_redis, make_message, make_result
+):
+    import asyncio
+
+    from taskiq_deduplication.middleware import DEDUP_TTL_LABEL
+
+    # 1s TTL with a sub-second heartbeat: without refresh the lock would expire.
+    mw.heartbeat_interval = 0.2
+    msg = make_message(labels={DEDUP_TTL_LABEL: 1})
+    await mw.pre_send(msg)
+    key = mw._build_deduplication_key(msg)
+    await mw.pre_execute(msg)
+    try:
+        # outlive the original TTL; the heartbeat should keep the lock present
+        await asyncio.sleep(1.5)
+        assert await real_redis.exists(key)
+        with pytest.raises(DuplicateTaskError):
+            await mw.pre_send(make_message(labels={DEDUP_TTL_LABEL: 1}))
+    finally:
+        await mw.post_execute(msg, make_result())
+    assert not await real_redis.exists(key)
+
+
+@pytest.mark.integration
+async def test_explicit_key_end_to_end(mw, real_redis, make_message, make_result):
+    from taskiq_deduplication.middleware import DEDUP_EXPLICIT_KEY_LABEL
+
+    msg = make_message(labels={DEDUP_EXPLICIT_KEY_LABEL: "my-lock"})
+    await mw.pre_send(msg)
+    assert await real_redis.exists("taskiq:deduplication:my-lock")
+
+    with pytest.raises(DuplicateTaskError):
+        await mw.pre_send(
+            make_message(
+                kwargs={"different": "kwargs"},
+                labels={DEDUP_EXPLICIT_KEY_LABEL: "my-lock"},
+            )
+        )
+
+    await mw.post_execute(msg, make_result())
+    assert not await real_redis.exists("taskiq:deduplication:my-lock")

tests/test_middleware.py

@@ -1,6 +1,7 @@
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from pydantic import RedisDsn, TypeAdapter
 
 from taskiq_deduplication import DuplicateTaskError, RedisDeduplicationMiddleware
 from taskiq_deduplication.middleware import (
@@ -8,6 +9,7 @@ from taskiq_deduplication.middleware import (
     DEDUP_KEY_FIELDS_LABEL,
     DEDUP_LABEL,
     DEDUP_TTL_LABEL,
+    SEND_GRACE_TTL,
 )
 
 
@@ -85,6 +87,27 @@ class TestDefaultBuildDeduplicationKey:
             m1
         ) == middleware._build_deduplication_key(m2)
 
+    def test_different_args_different_key(self, middleware, make_message):
+        m1 = make_message(args=["a"])
+        m2 = make_message(args=["b"])
+        assert middleware._build_deduplication_key(
+            m1
+        ) != middleware._build_deduplication_key(m2)
+
+    def test_arg_order_matters(self, middleware, make_message):
+        m1 = make_message(args=["a", "b"])
+        m2 = make_message(args=["b", "a"])
+        assert middleware._build_deduplication_key(
+            m1
+        ) != middleware._build_deduplication_key(m2)
+
+    def test_key_fields_ignores_args(self, middleware, make_message):
+        m1 = make_message(args=["a"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
+        m2 = make_message(args=["b"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
     def test_key_prefix_in_output(self, make_message):
         mw = RedisDeduplicationMiddleware(
             redis_url="redis://localhost", key_prefix="myapp:locks"
@@ -92,31 +115,93 @@ class TestDefaultBuildDeduplicationKey:
         mw._redis = None
         m = make_message()
         key = mw._build_deduplication_key(m)
-        assert key.startswith("myapp:locks:")
+        assert key is not None and key.startswith("myapp:locks:")
+
+    def test_empty_kwargs_produces_consistent_key(self, middleware, make_message):
+        m1 = make_message(kwargs={})
+        m2 = make_message(kwargs={})
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
+    def test_key_fields_empty_list_ignores_all_kwargs(self, middleware, make_message):
+        m1 = make_message(kwargs={"a": 1}, labels={DEDUP_KEY_FIELDS_LABEL: []})
+        m2 = make_message(kwargs={"a": 999}, labels={DEDUP_KEY_FIELDS_LABEL: []})
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
+    def test_key_fields_absent_from_kwargs_are_ignored(self, middleware, make_message):
+        m1 = make_message(
+            kwargs={"order_id": 1}, labels={DEDUP_KEY_FIELDS_LABEL: ["user_id"]}
+        )
+        m2 = make_message(
+            kwargs={"order_id": 999}, labels={DEDUP_KEY_FIELDS_LABEL: ["user_id"]}
+        )
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
+    def test_explicit_key_takes_precedence_over_key_fields(
+        self, middleware, make_message
+    ):
+        m = make_message(
+            kwargs={"a": 1},
+            labels={DEDUP_EXPLICIT_KEY_LABEL: "my-lock", DEDUP_KEY_FIELDS_LABEL: ["a"]},
+        )
+        assert middleware._build_deduplication_key(m) == "taskiq:deduplication:my-lock"
+
+    def test_non_serializable_kwargs_returns_none(self, middleware, make_message):
+        m = make_message(kwargs={"dt": object()})
+        assert middleware._build_deduplication_key(m) is None
 
 
 class TestPreSend:
-    @pytest.mark.anyio
     async def test_first_send_passes(self, middleware, make_message):
         msg = make_message()
         result = await middleware.pre_send(msg)
         assert result is msg
 
-    @pytest.mark.anyio
     async def test_duplicate_raises(self, middleware, make_message):
         msg = make_message()
         await middleware.pre_send(msg)
         with pytest.raises(DuplicateTaskError):
             await middleware.pre_send(make_message())
 
-    @pytest.mark.anyio
+    async def test_duplicate_error_carries_structured_attributes(
+        self, middleware, make_message
+    ):
+        holder = make_message(task_id="holder-task")
+        await middleware.pre_send(holder)
+        key = middleware._build_deduplication_key(holder)
+        with pytest.raises(DuplicateTaskError) as exc_info:
+            await middleware.pre_send(make_message(task_id="loser-task"))
+        err = exc_info.value
+        assert err.task_name == "my_task"
+        assert err.key == key
+        assert err.holder_task_id == "holder-task"
+        assert key in str(err)
+
+    async def test_duplicate_error_holder_none_when_lock_released_in_race(
+        self, make_message
+    ):
+        # The lock is released between the failed SET NX and the GET lookup, so
+        # GET returns None and holder_task_id is left unset.
+        redis = AsyncMock()
+        redis.set.return_value = False
+        redis.get.return_value = None
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        mw._redis = redis
+        with pytest.raises(DuplicateTaskError) as exc_info:
+            await mw.pre_send(make_message())
+        assert exc_info.value.holder_task_id is None
+
     async def test_deduplication_disabled_label(self, middleware, make_message):
         msg1 = make_message(labels={DEDUP_LABEL: False})
         msg2 = make_message(labels={DEDUP_LABEL: False})
         await middleware.pre_send(msg1)
         await middleware.pre_send(msg2)  # should not raise
 
-    @pytest.mark.anyio
     async def test_deduplication_disabled_by_default_init(
         self, fake_redis, make_message
     ):
@@ -127,22 +212,151 @@ class TestPreSend:
         await mw.pre_send(make_message())
         await mw.pre_send(make_message())  # should not raise
 
-    @pytest.mark.anyio
     async def test_ttl_applied(self, middleware, fake_redis, make_message):
         msg = make_message(labels={DEDUP_TTL_LABEL: 42})
         await middleware.pre_send(msg)
+        await middleware.post_send(msg)
         key = middleware._build_deduplication_key(msg)
         ttl = await fake_redis.ttl(key)
-        assert 0 < ttl <= 42
+        assert SEND_GRACE_TTL < ttl <= 42
 
-    @pytest.mark.anyio
     async def test_different_kwargs_both_pass(self, middleware, make_message):
         await middleware.pre_send(make_message(kwargs={"x": 1}))
         await middleware.pre_send(make_message(kwargs={"x": 2}))
 
+    async def test_non_serializable_kwargs_skips_deduplication(
+        self, middleware, make_message
+    ):
+        msg1 = make_message(kwargs={"dt": object()})
+        msg2 = make_message(kwargs={"dt": object()})
+        await middleware.pre_send(msg1)
+        await middleware.pre_send(msg2)  # should not raise
+
+    async def test_default_ttl_applied(self, fake_redis, make_message):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost", default_ttl=77)
+        mw._redis = fake_redis
+        msg = make_message()
+        await mw.pre_send(msg)
+        await mw.post_send(msg)
+        key = mw._build_deduplication_key(msg)
+        ttl = await fake_redis.ttl(key)
+        assert SEND_GRACE_TTL < ttl <= 77
+
+
+class TestPostSend:
+    async def test_pre_send_only_uses_grace_ttl(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message(labels={DEDUP_TTL_LABEL: 300})
+        await middleware.pre_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        assert 0 < await fake_redis.ttl(key) <= SEND_GRACE_TTL
+
+    async def test_post_send_extends_to_full_ttl(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message(labels={DEDUP_TTL_LABEL: 300})
+        await middleware.pre_send(msg)
+        await middleware.post_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        assert await fake_redis.ttl(key) > SEND_GRACE_TTL
+
+    async def test_post_send_does_not_extend_short_ttl(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message(labels={DEDUP_TTL_LABEL: 5})
+        await middleware.pre_send(msg)
+        await middleware.post_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        assert 0 < await fake_redis.ttl(key) <= 5
+
+    async def test_post_send_does_not_resurrect_released_lock(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        msg = make_message(labels={DEDUP_TTL_LABEL: 300})
+        await middleware.pre_send(msg)
+        await middleware.post_execute(msg, make_result())
+        await middleware.post_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        assert not await fake_redis.exists(key)
+
+    async def test_post_send_survives_redis_error(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message(labels={DEDUP_TTL_LABEL: 300})
+        await middleware.pre_send(msg)
+        middleware._refresh_script = AsyncMock(side_effect=ConnectionError("boom"))
+        await middleware.post_send(msg)  # should not raise
+        key = middleware._build_deduplication_key(msg)
+        assert 0 < await fake_redis.ttl(key) <= SEND_GRACE_TTL
+
+    async def test_post_send_deduplication_disabled_noop(
+        self, fake_redis, make_message
+    ):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", default_deduplication=False
+        )
+        mw._redis = fake_redis
+        msg = make_message()
+        await mw.pre_send(msg)
+        await mw.post_send(msg)  # should not raise
+
+
+class TestFailOpen:
+    @staticmethod
+    def _broken_middleware(fail_open):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", fail_open=fail_open
+        )
+        mw._redis = MagicMock()
+        mw._redis.set = AsyncMock(side_effect=ConnectionError("redis is down"))
+        return mw
+
+    async def test_redis_error_raises_by_default(self, make_message):
+        mw = self._broken_middleware(fail_open=False)
+        with pytest.raises(ConnectionError):
+            await mw.pre_send(make_message())
+
+    async def test_redis_error_lets_task_through_when_enabled(
+        self, make_message, caplog
+    ):
+        mw = self._broken_middleware(fail_open=True)
+        msg = make_message()
+        with caplog.at_level("WARNING"):
+            assert await mw.pre_send(msg) is msg
+        assert "fail_open" in caplog.text
+
+    async def test_duplicates_still_raise_when_enabled(self, fake_redis, make_message):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost", fail_open=True)
+        mw._redis = fake_redis
+        await mw.pre_send(make_message())
+        with pytest.raises(DuplicateTaskError):
+            await mw.pre_send(make_message())
+
+    async def test_fail_open_leaves_nothing_to_clean_up(
+        self, make_message, make_result
+    ):
+        mw = self._broken_middleware(fail_open=True)
+        msg = make_message()
+        await mw.pre_send(msg)
+        # None of the downstream hooks may touch the lock that was never taken.
+        await mw.post_send(msg)
+        await mw.pre_execute(msg)
+        await mw.post_execute(msg, make_result())
+        assert mw._heartbeats == {}
+
+    async def test_release_error_does_not_propagate(
+        self, middleware, make_message, make_result
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        middleware._release_script = AsyncMock(side_effect=ConnectionError("boom"))
+        # The task already ran: raising here would lose its result.
+        await middleware.post_execute(msg, make_result())
+        await middleware.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+
 
 class TestPostExecute:
-    @pytest.mark.anyio
     async def test_releases_lock(
         self, middleware, fake_redis, make_message, make_result
     ):
@@ -154,7 +368,6 @@ class TestPostExecute:
         await middleware.post_execute(msg, make_result())
         assert not await fake_redis.exists(key)
 
-    @pytest.mark.anyio
     async def test_deduplication_disabled_noop(
         self, middleware, fake_redis, make_message, make_result
     ):
@@ -166,9 +379,16 @@ class TestPostExecute:
         await middleware.post_execute(disabled_msg, make_result())
         assert await fake_redis.exists(key)
 
+    async def test_post_execute_after_ttl_expiry_is_safe(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        await fake_redis.delete(middleware._build_deduplication_key(msg))
+        await middleware.post_execute(msg, make_result())
+
 
 class TestOnError:
-    @pytest.mark.anyio
     async def test_releases_lock_on_error(
         self, middleware, fake_redis, make_message, make_result
     ):
@@ -180,7 +400,6 @@ class TestOnError:
         await middleware.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
         assert not await fake_redis.exists(key)
 
-    @pytest.mark.anyio
     async def test_deduplication_disabled_noop(
         self, middleware, fake_redis, make_message, make_result
     ):
@@ -195,8 +414,69 @@ class TestOnError:
         assert await fake_redis.exists(key)
 
 
+class TestSenderWorkerConfigMismatch:
+    async def test_post_execute_releases_lock_despite_worker_disabled_default(
+        self, fake_redis, make_message, make_result
+    ):
+        sender_mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", default_deduplication=True
+        )
+        sender_mw._redis = fake_redis
+
+        worker_mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", default_deduplication=False
+        )
+        worker_mw._redis = fake_redis
+
+        msg = make_message()
+        await sender_mw.pre_send(msg)
+        key = sender_mw._build_deduplication_key(msg)
+        assert await fake_redis.exists(key)
+
+        await worker_mw.post_execute(msg, make_result())
+        assert not await fake_redis.exists(key)
+
+    async def test_on_error_releases_lock_despite_worker_disabled_default(
+        self, fake_redis, make_message, make_result
+    ):
+        sender_mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", default_deduplication=True
+        )
+        sender_mw._redis = fake_redis
+
+        worker_mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", default_deduplication=False
+        )
+        worker_mw._redis = fake_redis
+
+        msg = make_message()
+        await sender_mw.pre_send(msg)
+        key = sender_mw._build_deduplication_key(msg)
+        assert await fake_redis.exists(key)
+
+        await worker_mw.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+        assert not await fake_redis.exists(key)
+
+
+class TestRedispatchAfterRelease:
+    async def test_redispatch_after_post_execute(
+        self, middleware, make_message, make_result
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        await middleware.post_execute(msg, make_result())
+        await middleware.pre_send(make_message())
+
+    async def test_redispatch_after_on_error(
+        self, middleware, make_message, make_result
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        await middleware.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+        await middleware.pre_send(make_message())
+
+
 class TestAtomicRelease:
-    @pytest.mark.anyio
     async def test_only_owner_can_release(self, middleware, fake_redis, make_message):
         owner_msg = make_message(task_id="owner-task")
         key = middleware._build_deduplication_key(owner_msg)
@@ -209,7 +489,6 @@ class TestAtomicRelease:
         await middleware._release_if_owned(key, "owner-task")
         assert not await fake_redis.exists(key)
 
-    @pytest.mark.anyio
     async def test_release_missing_key_is_noop(self, middleware, fake_redis):
         await middleware._release_if_owned(
             "taskiq:deduplication:nonexistent", "some-task"
@@ -217,18 +496,28 @@ class TestAtomicRelease:
 
 
 class TestLifecycle:
-    @pytest.mark.anyio
     async def test_startup_creates_redis_client(self):
         mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
         assert mw._redis is None
         with patch("redis.asyncio.Redis.from_url") as mock_from_url:
             mock_client = AsyncMock()
+            mock_client.register_script = MagicMock()
             mock_from_url.return_value = mock_client
             await mw.startup()
             mock_from_url.assert_called_once_with("redis://localhost")
             assert mw._redis is mock_client
 
-    @pytest.mark.anyio
+    async def test_startup_accepts_redis_dsn(self):
+        dsn = TypeAdapter(RedisDsn).validate_python("redis://localhost:6379/0")
+        mw = RedisDeduplicationMiddleware(redis_url=dsn)
+        with patch("redis.asyncio.Redis.from_url") as mock_from_url:
+            mock_client = AsyncMock()
+            mock_client.register_script = MagicMock()
+            mock_from_url.return_value = mock_client
+            await mw.startup()
+            mock_from_url.assert_called_once_with("redis://localhost:6379/0")
+            assert mw._redis is mock_client
+
     async def test_shutdown_closes_redis_client(self):
         mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
         mock_client = AsyncMock()
@@ -236,7 +525,513 @@ class TestLifecycle:
         await mw.shutdown()
         mock_client.aclose.assert_called_once()
 
-    @pytest.mark.anyio
     async def test_shutdown_without_startup_is_safe(self):
         mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
         await mw.shutdown()
+
+    async def test_pre_send_without_startup_raises_runtime_error(self, make_message):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        with pytest.raises(RuntimeError, match="startup"):
+            await mw.pre_send(make_message())
+
+
+class TestStartupRetry:
+    async def test_startup_succeeds_after_retries(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=3,
+            startup_retry_delay=0.01,
+        )
+        with patch("redis.asyncio.Redis.from_url") as mock_from_url:
+            mock_client = AsyncMock()
+            mock_client.ping.side_effect = [
+                ConnectionError("fail"),
+                ConnectionError("fail"),
+                None,
+            ]
+            mock_client.register_script = MagicMock()
+            mock_from_url.return_value = mock_client
+            await mw.startup()
+            assert mw._redis is mock_client
+            assert mock_client.ping.call_count == 3
+
+    async def test_startup_raises_after_all_retries_exhausted(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=2,
+            startup_retry_delay=0.01,
+        )
+        with patch("redis.asyncio.Redis.from_url") as mock_from_url:
+            mock_client = AsyncMock()
+            mock_client.ping.side_effect = ConnectionError("refused")
+            mock_from_url.return_value = mock_client
+            with pytest.raises(ConnectionError, match="2 attempts"):
+                await mw.startup()
+            assert mock_client.ping.call_count == 2
+
+    async def test_startup_no_retry_on_first_success(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=3,
+            startup_retry_delay=0.01,
+        )
+        with patch("redis.asyncio.Redis.from_url") as mock_from_url:
+            mock_client = AsyncMock()
+            mock_client.register_script = MagicMock()
+            mock_from_url.return_value = mock_client
+            await mw.startup()
+            mock_client.ping.assert_called_once()
+
+    async def test_startup_retry_delay_exponential(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=3,
+            startup_retry_delay=0.01,
+        )
+        with (
+            patch("redis.asyncio.Redis.from_url") as mock_from_url,
+            patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
+        ):
+            mock_client = AsyncMock()
+            mock_client.ping.side_effect = [
+                ConnectionError("fail"),
+                ConnectionError("fail"),
+                None,
+            ]
+            mock_client.register_script = MagicMock()
+            mock_from_url.return_value = mock_client
+            await mw.startup()
+            assert mock_sleep.call_count == 2
+            assert mock_sleep.call_args_list[0].args[0] == 0.01
+            assert mock_sleep.call_args_list[1].args[0] == 0.02
+
+    async def test_startup_closes_failed_clients(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=3,
+            startup_retry_delay=0.01,
+        )
+        failed1, failed2, good = AsyncMock(), AsyncMock(), AsyncMock()
+        failed1.ping.side_effect = ConnectionError("fail")
+        failed2.ping.side_effect = ConnectionError("fail")
+        good.register_script = MagicMock()
+
+        with patch(
+            "redis.asyncio.Redis.from_url", side_effect=[failed1, failed2, good]
+        ):
+            await mw.startup()
+
+        failed1.aclose.assert_called_once()
+        failed2.aclose.assert_called_once()
+        good.aclose.assert_not_called()
+        assert mw._redis is good
+
+    async def test_startup_closes_client_when_all_retries_exhausted(self):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost",
+            startup_retries=2,
+            startup_retry_delay=0.01,
+        )
+        failed1, failed2 = AsyncMock(), AsyncMock()
+        failed1.ping.side_effect = ConnectionError("fail")
+        failed2.ping.side_effect = ConnectionError("fail")
+
+        with (
+            patch("redis.asyncio.Redis.from_url", side_effect=[failed1, failed2]),
+            pytest.raises(ConnectionError),
+        ):
+            await mw.startup()
+
+        failed1.aclose.assert_called_once()
+        failed2.aclose.assert_called_once()
+
+
+class TestLabelTypeParsing:
+    async def test_bool_label_false_disables_dedup(self, fake_redis, make_message):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        mw._redis = fake_redis
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: False}))
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: False}))
+
+    async def test_bool_label_true_enables_dedup(self, middleware, make_message):
+        msg = make_message(labels={DEDUP_LABEL: True})
+        await middleware.pre_send(msg)
+        with pytest.raises(DuplicateTaskError):
+            await middleware.pre_send(make_message(labels={DEDUP_LABEL: True}))
+
+    async def test_string_bool_label_true_enables_dedup(self, middleware, make_message):
+        # taskiq's prepare_label() stringifies True → "True" before pre_send runs
+        msg = make_message(labels={DEDUP_LABEL: "True"})
+        await middleware.pre_send(msg)
+        with pytest.raises(DuplicateTaskError):
+            await middleware.pre_send(make_message(labels={DEDUP_LABEL: "True"}))
+
+    async def test_string_bool_label_false_disables_dedup(
+        self, fake_redis, make_message
+    ):
+        # taskiq's prepare_label() stringifies False → "False" before pre_send runs
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        mw._redis = fake_redis
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: "False"}))
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: "False"}))
+
+    async def test_key_fields_list_parsed_correctly(self, middleware, make_message):
+        m1 = make_message(
+            kwargs={"a": 1, "b": 2, "c": 3},
+            labels={DEDUP_KEY_FIELDS_LABEL: ["a", "b"]},
+        )
+        m2 = make_message(
+            kwargs={"a": 1, "b": 2, "c": 999},
+            labels={DEDUP_KEY_FIELDS_LABEL: ["a", "b"]},
+        )
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
+    async def test_key_fields_non_list_ignored(self, middleware, make_message):
+        m = make_message(
+            kwargs={"a": 1},
+            labels={DEDUP_KEY_FIELDS_LABEL: "not-a-list"},
+        )
+        key = middleware._build_deduplication_key(m)
+        assert key is not None
+
+    async def test_invalid_bool_label_warns_and_uses_default(
+        self, middleware, make_message, caplog
+    ):
+        import logging
+
+        msg = make_message(labels={DEDUP_LABEL: "yes"})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            await middleware.pre_send(msg)
+        assert any("yes" in r.message for r in caplog.records)
+
+    async def test_string_true_lowercase_enables_dedup(self, middleware, make_message):
+        msg = make_message(labels={DEDUP_LABEL: "true"})
+        await middleware.pre_send(msg)
+        with pytest.raises(DuplicateTaskError):
+            await middleware.pre_send(make_message(labels={DEDUP_LABEL: "true"}))
+
+    async def test_string_false_lowercase_disables_dedup(
+        self, fake_redis, make_message
+    ):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        mw._redis = fake_redis
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: "false"}))
+        await mw.pre_send(make_message(labels={DEDUP_LABEL: "false"}))
+
+    async def test_invalid_key_fields_label_warns_and_falls_back_to_all_kwargs(
+        self, middleware, make_message, caplog
+    ):
+        import logging
+
+        msg = make_message(kwargs={"a": 1}, labels={DEDUP_KEY_FIELDS_LABEL: "user_id"})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            key = middleware._build_deduplication_key(msg)
+        assert any("user_id" in r.message for r in caplog.records)
+        # falls back to full-kwargs fingerprint — key must still be produced
+        assert key is not None
+
+    async def test_key_fields_string_parses_to_non_list_warns(
+        self, middleware, make_message, caplog
+    ):
+        import logging
+
+        # ast.literal_eval succeeds but returns a dict, not a list
+        msg = make_message(kwargs={"a": 1}, labels={DEDUP_KEY_FIELDS_LABEL: "{'a': 1}"})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            key = middleware._build_deduplication_key(msg)
+        assert any("{'a': 1}" in r.message for r in caplog.records)
+        assert key is not None
+
+    async def test_missing_key_fields_warns(self, middleware, make_message, caplog):
+        import logging
+
+        msg = make_message(
+            kwargs={"order_id": 1},
+            labels={DEDUP_KEY_FIELDS_LABEL: ["user_id", "order_id"]},
+        )
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            key = middleware._build_deduplication_key(msg)
+        assert any("user_id" in r.message for r in caplog.records)
+        # the field that is present must not be reported as missing
+        assert not any("order_id" in r.message for r in caplog.records)
+        assert key is not None
+
+    async def test_present_key_fields_do_not_warn(
+        self, middleware, make_message, caplog
+    ):
+        import logging
+
+        msg = make_message(
+            kwargs={"user_id": 1, "order_id": 2},
+            labels={DEDUP_KEY_FIELDS_LABEL: ["user_id"]},
+        )
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            middleware._build_deduplication_key(msg)
+        assert not any("absent from kwargs" in r.message for r in caplog.records)
+
+    def test_stringified_key_fields_parsed_correctly(self, middleware, make_message):
+        # taskiq's prepare_label() stringifies ["a", "b"] → "['a', 'b']" before pre_send
+        m1 = make_message(
+            kwargs={"a": 1, "b": 2, "c": 3},
+            labels={DEDUP_KEY_FIELDS_LABEL: "['a', 'b']"},
+        )
+        m2 = make_message(
+            kwargs={"a": 1, "b": 2, "c": 999},
+            labels={DEDUP_KEY_FIELDS_LABEL: "['a', 'b']"},
+        )
+        assert middleware._build_deduplication_key(
+            m1
+        ) == middleware._build_deduplication_key(m2)
+
+    def test_stringified_key_fields_different_included_fields(
+        self, middleware, make_message
+    ):
+        m1 = make_message(
+            kwargs={"a": 1, "b": 2},
+            labels={DEDUP_KEY_FIELDS_LABEL: "['a']"},
+        )
+        m2 = make_message(
+            kwargs={"a": 99, "b": 2},
+            labels={DEDUP_KEY_FIELDS_LABEL: "['a']"},
+        )
+        assert middleware._build_deduplication_key(
+            m1
+        ) != middleware._build_deduplication_key(m2)
+
+    async def test_invalid_ttl_string_warns_and_uses_default(
+        self, middleware, fake_redis, make_message, caplog
+    ):
+        import logging
+
+        msg = make_message(labels={DEDUP_TTL_LABEL: "oops"})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            await middleware.pre_send(msg)
+        assert any("oops" in r.message for r in caplog.records)
+        key = middleware._build_deduplication_key(msg)
+        ttl = await fake_redis.ttl(key)
+        assert 0 < ttl <= middleware.default_ttl
+
+    async def test_invalid_ttl_none_warns_and_uses_default(
+        self, middleware, fake_redis, make_message, caplog
+    ):
+        import logging
+
+        msg = make_message(labels={DEDUP_TTL_LABEL: None})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            await middleware.pre_send(msg)
+        assert any("None" in r.message for r in caplog.records)
+        key = middleware._build_deduplication_key(msg)
+        ttl = await fake_redis.ttl(key)
+        assert 0 < ttl <= middleware.default_ttl
+
+
+class TestKeyCaching:
+    async def test_key_cached_during_pre_send(self, middleware, make_message):
+        from taskiq_deduplication.middleware import _CACHED_KEY_LABEL
+
+        msg = make_message()
+        await middleware.pre_send(msg)
+        assert _CACHED_KEY_LABEL in msg.labels
+
+    async def test_post_execute_uses_cached_key(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        from taskiq_deduplication.middleware import _CACHED_KEY_LABEL
+
+        msg = make_message()
+        await middleware.pre_send(msg)
+        cached_key = msg.labels.get(_CACHED_KEY_LABEL)
+        await middleware.post_execute(msg, make_result())
+        assert not await fake_redis.exists(cached_key)
+
+    async def test_on_error_uses_cached_key(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        from taskiq_deduplication.middleware import _CACHED_KEY_LABEL
+
+        msg = make_message()
+        await middleware.pre_send(msg)
+        cached_key = msg.labels.get(_CACHED_KEY_LABEL)
+        await middleware.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+        assert not await fake_redis.exists(cached_key)
+
+    async def test_cached_key_none_when_key_build_fails(self, middleware, make_message):
+        from taskiq_deduplication.middleware import _CACHED_KEY_LABEL
+
+        msg = make_message(kwargs={"dt": object()})
+        await middleware.pre_send(msg)
+        assert msg.labels[_CACHED_KEY_LABEL] is None
+
+    async def test_post_execute_noop_when_cached_key_is_none(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        msg = make_message(kwargs={"dt": object()})
+        await middleware.pre_send(msg)
+        await middleware.post_execute(msg, make_result())
+
+    async def test_on_error_noop_when_cached_key_is_none(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        msg = make_message(kwargs={"dt": object()})
+        await middleware.pre_send(msg)
+        await middleware.on_error(msg, make_result(is_err=True), RuntimeError("boom"))
+
+    async def test_release_if_owned_raises_without_redis(
+        self, middleware, make_message
+    ):
+        middleware._redis = None
+        with pytest.raises(RuntimeError, match="startup"):
+            await middleware._release_if_owned("some-key", "some-task")
+
+
+class TestTTLExpiry:
+    async def test_lock_expiry_admits_duplicate(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        # simulate TTL expiry by deleting the key
+        await fake_redis.delete(middleware._build_deduplication_key(msg))
+        # same fingerprint should now pass since lock is gone
+        await middleware.pre_send(make_message())
+
+
+class TestHeartbeat:
+    async def test_pre_execute_starts_heartbeat(self, middleware, make_message):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        await middleware.pre_execute(msg)
+        assert msg.task_id in middleware._heartbeats
+        await middleware._cancel_heartbeat(msg.task_id)
+
+    async def test_pre_execute_noop_when_heartbeat_disabled(
+        self, fake_redis, make_message
+    ):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", heartbeat=False
+        )
+        mw._redis = fake_redis
+        msg = make_message()
+        await mw.pre_send(msg)
+        await mw.pre_execute(msg)
+        assert msg.task_id not in mw._heartbeats
+
+    async def test_pre_execute_noop_without_cached_key(self, middleware, make_message):
+        # deduplication disabled -> pre_send never caches a key
+        msg = make_message(labels={DEDUP_LABEL: False})
+        await middleware.pre_send(msg)
+        await middleware.pre_execute(msg)
+        assert msg.task_id not in middleware._heartbeats
+
+    async def test_heartbeat_refreshes_ttl(self, middleware, fake_redis, make_message):
+        import asyncio
+
+        # short ttl, tiny heartbeat interval so the lock would expire without refresh
+        middleware.heartbeat_interval = 0.05
+        msg = make_message(labels={DEDUP_TTL_LABEL: 1})
+        await middleware.pre_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        await middleware.pre_execute(msg)
+        try:
+            # let several heartbeats elapse — longer than the original 1s ttl
+            await asyncio.sleep(0.3)
+            assert await fake_redis.exists(key)
+            ttl = await fake_redis.ttl(key)
+            assert 0 < ttl <= 1
+        finally:
+            await middleware._cancel_heartbeat(msg.task_id)
+
+    async def test_release_lock_cancels_heartbeat(
+        self, middleware, fake_redis, make_message, make_result
+    ):
+        msg = make_message()
+        await middleware.pre_send(msg)
+        await middleware.pre_execute(msg)
+        assert msg.task_id in middleware._heartbeats
+        await middleware.post_execute(msg, make_result())
+        assert msg.task_id not in middleware._heartbeats
+        key = middleware._build_deduplication_key(msg)
+        assert not await fake_redis.exists(key)
+
+    async def test_heartbeat_stops_when_lock_lost(
+        self, middleware, fake_redis, make_message
+    ):
+        import asyncio
+
+        middleware.heartbeat_interval = 0.05
+        msg = make_message(labels={DEDUP_TTL_LABEL: 1})
+        await middleware.pre_send(msg)
+        key = middleware._build_deduplication_key(msg)
+        await middleware.pre_execute(msg)
+        # another task steals the key
+        await fake_redis.set(key, "other-task", ex=10)
+        await asyncio.sleep(0.15)
+        task = middleware._heartbeats.get(msg.task_id)
+        # heartbeat loop should have returned on its own
+        assert task is None or task.done()
+        await middleware._cancel_heartbeat(msg.task_id)
+
+    async def test_shutdown_cancels_heartbeats(self, fake_redis, make_message):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        mw._redis = fake_redis
+        msg = make_message()
+        await mw.pre_send(msg)
+        await mw.pre_execute(msg)
+        assert msg.task_id in mw._heartbeats
+        await mw.shutdown()
+        assert not mw._heartbeats
+
+    async def test_default_heartbeat_interval_is_third_of_ttl(self, middleware):
+        assert middleware._get_heartbeat_interval(300) == 100.0
+        assert middleware._get_heartbeat_interval(1) == 1.0
+
+    async def test_explicit_heartbeat_interval_overrides(self, fake_redis):
+        mw = RedisDeduplicationMiddleware(
+            redis_url="redis://localhost", heartbeat_interval=5.0
+        )
+        mw._redis = fake_redis
+        assert mw._get_heartbeat_interval(300) == 5.0
+
+    async def test_refresh_if_owned_raises_without_redis(self, middleware):
+        middleware._redis = None
+        with pytest.raises(RuntimeError, match="startup"):
+            await middleware._refresh_if_owned("some-key", "some-task", 60)
+
+    async def test_heartbeat_continues_after_refresh_error(
+        self, middleware, make_message, caplog
+    ):
+        import asyncio
+        import logging
+
+        middleware.heartbeat_interval = 0.02
+        # first refresh raises, subsequent ones succeed; the loop must survive
+        middleware._refresh_if_owned = AsyncMock(
+            side_effect=[ConnectionError("boom"), True, True, True]
+        )
+        msg = make_message()
+        await middleware.pre_send(msg)
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.middleware"):
+            await middleware.pre_execute(msg)
+            await asyncio.sleep(0.1)
+            task = middleware._heartbeats.get(msg.task_id)
+            # loop swallowed the error and kept running
+            assert task is not None and not task.done()
+            await middleware._cancel_heartbeat(msg.task_id)
+        assert any("Failed to refresh lock" in r.message for r in caplog.records)
+        assert middleware._refresh_if_owned.call_count >= 2
+
+
+class TestExplicitKeyEdgeCases:
+    def test_empty_string_key_produces_prefix_only_key(self, middleware, make_message):
+        m = make_message(labels={DEDUP_EXPLICIT_KEY_LABEL: ""})
+        key = middleware._build_deduplication_key(m)
+        assert key == "taskiq:deduplication:"
+
+    async def test_empty_string_key_acquires_lock(
+        self, middleware, fake_redis, make_message
+    ):
+        msg = make_message(labels={DEDUP_EXPLICIT_KEY_LABEL: ""})
+        await middleware.pre_send(msg)
+        assert await fake_redis.exists("taskiq:deduplication:")

zensical.toml

@@ -7,6 +7,7 @@ copyright = "Copyright © 2026 d3vyce"
 repo_url = "https://github.com/d3vyce/taskiq-deduplication"
 
 [project.theme]
+custom_dir = "docs/overrides"
 language = "en"
 features = [
     "announce.dismiss",