fix(prefect-redis): make socket timeouts reachable and publisher/reconnect safe

[Adisa-Shobi]

closes #22478

clear_cached_clients() evicted cached Redis clients from the dict without closing them, so a connection stuck in-use from a hung operation (e.g. a blocked XREADGROUP during an outage) leaked forever, eventually exhausting the pool's connection cap even after Redis recovered. This also fixes the two things that keep that recovery path from actually firing and from losing data once it does.

Details

• clear_cached_clients() now awaits client.aclose() for each evicted client instead of just dropping the cache entry, and accepts an optional client= argument so Consumer.run retires only the client that failed instead of force-closing every current-loop client (including healthy ones for other endpoints).
• socket_timeout/socket_connect_timeout default to 60.0/10.0 instead of None. With None, a dead socket never raises, so Consumer.run's except RedisError reconnect branch — the thing this PR fixes — is never reached in the first place.
• Publisher._publish_periodically() now survives a flush failure instead of dying silently, and _publish_current_batch() re-queues the unsent tail of a batch instead of dropping it. Without this, a finite timeout turns a hang into silent data loss on the next sustained outage.

Known and deliberate, left out of scope:

• RedisWorkerCleanupQueueSettings (cleanup_queue.py) has the same socket_timeout=None default and the same latent hang, minus the amplifier — it polls rather than blocking. Separate settings class, separate client path that never touches _client_cache.
• Every production call site of get_async_redis_client() passes no arguments, so today all cached clients share one cache key and per-client targeting is a no-op in practice. It matters once a caller passes an explicit url=/host=.

Validation:

• PYTHONPATH=src/integrations/prefect-redis .venv/bin/pytest src/integrations/prefect-redis/tests/test_client.py src/integrations/prefect-redis/tests/test_messaging.py
• PYTHONPATH=src/integrations/prefect-redis .venv/bin/ruff check src/integrations/prefect-redis/prefect_redis/ src/integrations/prefect-redis/tests/
• PYTHONPATH=src/integrations/prefect-redis .venv/bin/mypy src/integrations/prefect-redis/prefect_redis/client.py src/integrations/prefect-redis/prefect_redis/messaging.py
• Full suite: 191 passed, 3 skipped. Requires a live Redis on localhost:6379.

Checklist

• This pull request references any related issue by including "closes #22478"
• If this pull request adds or changes functionality, it includes tests or explains why tests are not needed.
• If this pull request changes user-facing behavior, it updates documentation or explains why documentation is not needed.
• If this pull request removes docs files, it includes redirect settings in mint.json.
• If this pull request adds functions or classes, it includes helpful docstrings.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 4 new potential issues.