feat: add fail_open mode so a Redis outage cannot block task dispatch (#85)

README.md

@@ -58,6 +58,7 @@ except DuplicateTaskError:
 - **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

@@ -58,6 +58,7 @@ except DuplicateTaskError:
 - **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/usage.md

@@ -25,6 +25,7 @@ broker = ListQueueBroker("redis://localhost:6379").with_middlewares(
 | `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(
@@ -81,6 +82,29 @@ RedisDeduplicationMiddleware(
 )
 ```
 
+## 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

src/taskiq_deduplication/middleware.py

@@ -74,6 +74,9 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             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__(
@@ -86,6 +89,7 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         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
@@ -95,6 +99,7 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         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
@@ -239,11 +244,25 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         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=min(ttl, SEND_GRACE_TTL), nx=True
-        )
+        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(
+                "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:
-            holder_task_id = await self._redis.get(key)
             if isinstance(holder_task_id, bytes):
                 holder_task_id = holder_task_id.decode()
             logger.warning(
@@ -342,7 +361,12 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         key = self._get_cached_key(message)
         if key is None:
             return
-        await self._release_if_owned(key, message.task_id)
+        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,

tests/test_middleware.py

@@ -302,6 +302,60 @@ class TestPostSend:
         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:
     async def test_releases_lock(
         self, middleware, fake_redis, make_message, make_result