Improve Redis connection pool recovery after Redis outages#22481
Improve Redis connection pool recovery after Redis outages#22481makhdoombilawal wants to merge 3 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR strengthens the prefect-redis integration’s ability to recover from temporary Redis outages by ensuring cached Redis clients are properly torn down and by broadening the consumer retry behavior to include socket-level failures.
Changes:
- Add explicit connection-pool disconnect + client close steps before clearing the cached Redis client map.
- Treat
OSErroras a retryable connection failure in the Redis consumer loop (in addition toRedisError). - Add regression tests for cached-client cleanup and consumer retry behavior during connection failures.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| src/integrations/prefect-redis/prefect_redis/client.py | Adds best-effort disconnect/close of cached clients before clearing the cache to avoid “stuck” broken connections. |
| src/integrations/prefect-redis/prefect_redis/messaging.py | Broadens retry handling to include OSError and tightens a few type hints to match server messaging protocols. |
| src/integrations/prefect-redis/tests/test_connection_recovery.py | New tests covering client cleanup and consumer retry on OSError/connection failures. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| import asyncio | ||
| from unittest.mock import AsyncMock, MagicMock, patch | ||
| import pytest | ||
|
|
||
| from prefect_redis.client import _client_cache, clear_cached_clients | ||
| from prefect_redis.messaging import Consumer | ||
| from prefect.server.utilities.messaging import StopConsumer |
| @pytest.mark.asyncio | ||
| async def test_clear_cached_clients_disconnects_pools(): | ||
| mock_client = MagicMock() | ||
| mock_client.connection_pool = AsyncMock() | ||
| mock_client.aclose = AsyncMock() | ||
|
|
||
| mock_loop = asyncio.get_running_loop() | ||
| cache_key = (MagicMock(), (), (), mock_loop) | ||
|
|
||
| _client_cache[cache_key] = mock_client |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 07b769351e
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| continue | ||
| try: | ||
| if hasattr(client, "connection_pool") and client.connection_pool: | ||
| await client.connection_pool.disconnect() |
There was a problem hiding this comment.
During recovery the cache remains populated while disconnect()/aclose() are awaited, so any concurrent publisher/consumer that calls get_async_redis_client() during this window will be handed the same stale client that is being disconnected instead of a fresh one. This regresses the recovery path under concurrent Redis users; snapshot the clients and clear _client_cache before awaiting shutdown so future lookups immediately create new connections.
Useful? React with 👍 / 👎.
…_clients, fix import order and trailing whitespace in tests
Summary
This PR improves Redis connection recovery in the
prefect-redisintegration after temporary Redis outages.Changes
OSError) failures in the consumer retry loop.Motivation
When Redis becomes unavailable, cached clients can remain in a broken state, preventing automatic recovery and requiring a process restart. These changes improve cleanup and retry behavior to make recovery more reliable.
Testing
prefect-redistest suite successfully.