Recover prefect-redis connection pool after a Redis outage

[devin-ai-integration]

closes #22478

This PR makes the prefect-redis shared async client recover after the Redis broker becomes unreachable, instead of permanently raising redis.exceptions.MaxConnectionsError: Too many connections until a process restart.

Root cause

The cached client's single ConnectionPool had several problems that combined into a permanent failure:

1. socket_timeout/socket_connect_timeout defaulted to None, so a blocking XREADGROUP on a dead socket hangs forever. The connection stays in _in_use_connections, never released. As reads pile up, the pool hits its default cap (100) and every op raises MaxConnectionsError — even after Redis is healthy again. The indefinite hang also means the consumer's except RedisError reconnect path never fires.
2. redis-py 8 defaults commands to 10 retries and holds the connection checked out across every attempt. On the direct Redis(...) path a dead broker can therefore occupy a pool slot for ~11 × socket_timeout before the consumer ever sees the error, while the Redis.from_url() path gets a different (near-zero) policy — so the two construction paths behaved inconsistently.
3. clear_cached_clients() only did _client_cache.clear(); the orphaned pool (with its stuck in-use connections) was never disconnected.

Changes

• RedisMessagingSettings: default socket_timeout=60.0 and socket_connect_timeout=10.0 (still overridable/None-able). A read against an unreachable broker now fails fast, so its connection is released and the reconnect path engages. socket_timeout is deliberately well above the default consumer block (1s) so idle blocking reads don't time out spuriously.
• Bounded, consistent retry policy applied to both the Redis(...) and Redis.from_url() construction paths (in get_async_redis_client and async_redis_from_settings) via a new retries setting (default 3). Only ConnectionError is retried — read timeouts are not retried, so a blocking read fails fast into the consumer's reconnect loop instead of pinning the pool slot for many multiples of the socket timeout.
• clear_cached_clients(): detaches the current-loop entries it will close (_client_cache.pop) before the first await, then disconnects (disconnect(inuse_connections=True)) and aclose()s only that detached snapshot. It no longer clears the cache wholesale, so a replacement client cached concurrently while it awaits survives (previously the freshly recovered pool could be orphaned). Clients bound to other event loops are left untouched.
• Consumer block/timeout warning: now treats BLOCK 0 as infinite (it out-lives any finite socket timeout, so it's flagged even though its numeric value is 0), and reads the effective socket_timeout off the client's connection kwargs — respecting URL query overrides rather than only the settings default. Moved into run() where the actual client is available.

Before/after (reproduced with a black-hole TCP proxy in front of Redis)

Before — after the broker recovers, the pool stays saturated:

[after recovery] max=100 in_use=100 available=0
ping after recovery STILL -> MaxConnectionsError: Too many connections
clear_cached_clients(); orphaned pool in_use still = 100

After — blocking reads fail fast, connections are reaped, and the client resumes:

reads done after timeout: 100/100
in_use after timeout: 0
orphaned pool in_use after clear: 0
RECOVERED: ping ok after outage

Relationship to #22457

#22457 (open, draft) reworks the client lifecycle (removes the global cache, adds managed_async_redis_client()) and closes the related FD-leak issue #22442. It does not change the socket-timeout defaults or the retry policy, so on its own it does not resolve this outage-recovery scenario (with socket_timeout=None the blocking read still never raises, so the managed client is never closed). The timeout/retry fixes here are required regardless of that refactor; the clear_cached_clients() change targets the current (still-cached) code on main.

Checklist

• References the related issue (closes #22478)
• Adds tests: socket-timeout defaults, bounded retry on all three construction paths, host/port + from_url outage→recovery regression tests, a consumer read-timeout recovery test, clear_cached_clients() disconnect/close + concurrent-replacement race, and the block/timeout warning (incl. BLOCK 0 and URL-override cases)
• User-facing behavior change (default socket timeouts, bounded retries) documented in the setting descriptions
• Adds/updates docstrings

Link to Devin session: https://app.devin.ai/sessions/f4e8b325760a40da9c792f28edf69a21
Requested by: @desertaxle

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring