diff --git a/README.md b/README.md index a40d509..2baac96 100644 --- a/README.md +++ b/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 diff --git a/docs/index.md b/docs/index.md index a40d509..2baac96 100644 --- a/docs/index.md +++ b/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 diff --git a/docs/usage.md b/docs/usage.md index f582be8..37363c3 100644 --- a/docs/usage.md +++ b/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 diff --git a/src/taskiq_deduplication/middleware.py b/src/taskiq_deduplication/middleware.py index 8a90a3d..850382b 100644 --- a/src/taskiq_deduplication/middleware.py +++ b/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, diff --git a/tests/test_middleware.py b/tests/test_middleware.py index 7789cbb..286afda 100644 --- a/tests/test_middleware.py +++ b/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