feat: add startup retry with exponential backoff for Redis connection (#20)

README.md

@@ -50,12 +50,13 @@ 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.
+- **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.
 - **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.
 - **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.
 
 ## License
 

docs/index.md

@@ -50,12 +50,13 @@ 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.
+- **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.
 - **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.
 - **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.
 
 ## License
 

docs/usage.md

@@ -21,6 +21,8 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
 | `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). |
 
 ```python
 broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
@@ -29,10 +31,29 @@ 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,
     ),
 )
 ```
 
+## 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,
+)
+```
+
 ## How it works
 
 When a task is dispatched, the middleware acquires a Redis lock keyed on the task's

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "taskiq-deduplication"
-version = "1.0.1"
-description = "Production-ready utilities for FastAPI applications"
+version = "1.0.2"
+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",

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.1"
+__version__ = "1.0.2"
 
 __all__ = [
     "DuplicateTaskError",

src/taskiq_deduplication/middleware.py

@@ -1,7 +1,8 @@
+import asyncio
 import hashlib
 import json
 import logging
-from typing import Any
+from typing import Any, Awaitable, cast
 
 from redis.asyncio import Redis
 from taskiq import TaskiqMessage, TaskiqResult
@@ -42,21 +43,49 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         default_deduplication: bool = True,
         default_ttl: int = 300,
         key_prefix: str = "taskiq:deduplication",
+        startup_retries: int = 3,
+        startup_retry_delay: float = 1.0,
     ) -> 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._redis: Redis | None = None
 
     async def startup(self) -> None:
+        last_error: BaseException | None = None
+        for attempt in range(self.startup_retries):
+            try:
                 self._redis = Redis.from_url(self.redis_url)
+                await cast(Awaitable[bool], self._redis.ping())
+                return
+            except Exception as exc:
+                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:
         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}"
@@ -67,10 +96,13 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             if key_fields is not None
             else message.kwargs
         )
+        try:
             payload = json.dumps(
                 {"task": message.task_name, "kwargs": kwargs},
                 sort_keys=True,
             )
+        except TypeError:
+            return None
         fingerprint = hashlib.sha256(payload.encode()).hexdigest()[:16]
         return f"{self.key_prefix}:{fingerprint}"
 
@@ -81,7 +113,10 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         return int(labels.get(DEDUP_TTL_LABEL, self.default_ttl))
 
     async def _release_if_owned(self, key: str, task_id: str) -> None:
-        assert self._redis is not None
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
         released = await check_and_delete(self._redis, key, task_id)
         if released:
             logger.debug("Released lock %s", key)
@@ -92,8 +127,18 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         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)
+        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)
@@ -118,9 +163,10 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
     ) -> None:
         if not self._is_enabled(message.labels):
             return
-        await self._release_if_owned(
-            self._build_deduplication_key(message), message.task_id
-        )
+        key = self._build_deduplication_key(message)
+        if key is None:
+            return
+        await self._release_if_owned(key, message.task_id)
 
     async def on_error(
         self,
@@ -130,6 +176,7 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
     ) -> None:
         if not self._is_enabled(message.labels):
             return
-        await self._release_if_owned(
-            self._build_deduplication_key(message), message.task_id
-        )
+        key = self._build_deduplication_key(message)
+        if key is None:
+            return
+        await self._release_if_owned(key, message.task_id)

src/taskiq_deduplication/utils.py

@@ -1,18 +1,26 @@
-from collections.abc import Awaitable
-from typing import cast
-
 from redis.asyncio import Redis
+from redis.commands.core import AsyncScript
+
+RELEASE_LUA_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(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
-    """
+    """Delete *key* only if its value equals *owner*.
 
-    released = await cast(Awaitable[int], redis.eval(release_script, 1, key, owner))
+    Args:
+        redis: Async Redis client.
+        key: Lock key to delete.
+        owner: Expected value of the key (task_id).
+
+    Returns:
+        True if the key was deleted, False otherwise.
+    """
+    script: AsyncScript = redis.register_script(RELEASE_LUA_SCRIPT)
+    released: int = await script(keys=[key], args=[owner])
     return bool(released)

tests/test_middleware.py

@@ -92,7 +92,45 @@ 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:
@@ -140,6 +178,25 @@ class TestPreSend:
         await middleware.pre_send(make_message(kwargs={"x": 1}))
         await middleware.pre_send(make_message(kwargs={"x": 2}))
 
+    @pytest.mark.anyio
+    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
+
+    @pytest.mark.anyio
+    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)
+        key = mw._build_deduplication_key(msg)
+        ttl = await fake_redis.ttl(key)
+        assert 0 < ttl <= 77
+
 
 class TestPostExecute:
     @pytest.mark.anyio
@@ -166,6 +223,15 @@ class TestPostExecute:
         await middleware.post_execute(disabled_msg, make_result())
         assert await fake_redis.exists(key)
 
+    @pytest.mark.anyio
+    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
@@ -195,6 +261,26 @@ class TestOnError:
         assert await fake_redis.exists(key)
 
 
+class TestRedispatchAfterRelease:
+    @pytest.mark.anyio
+    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())
+
+    @pytest.mark.anyio
+    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):
@@ -240,3 +326,81 @@ class TestLifecycle:
     async def test_shutdown_without_startup_is_safe(self):
         mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
         await mw.shutdown()
+
+    @pytest.mark.anyio
+    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:
+    @pytest.mark.anyio
+    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_from_url.return_value = mock_client
+            await mw.startup()
+            assert mw._redis is mock_client
+            assert mock_client.ping.call_count == 3
+
+    @pytest.mark.anyio
+    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
+
+    @pytest.mark.anyio
+    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_from_url.return_value = mock_client
+            await mw.startup()
+            mock_client.ping.assert_called_once()
+
+    @pytest.mark.anyio
+    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_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

uv.lock

@@ -1468,7 +1468,7 @@ wheels = [
 
 [[package]]
 name = "taskiq-deduplication"
-version = "1.0.1"
+version = "1.0.2"
 source = { editable = "." }
 dependencies = [
     { name = "redis" },