fix: bound the queue read so a dead Redis socket cannot park a worker forever

[adhikjoshi]

What broke

On 2026-08-15 ~04:05 UTC the TCP connections from the GPU fleet to the Redis box were silently dropped. Seven server_types stopped consuming within fifteen minutes of each other:

server_type	last success (UTC)
z_image_turbo	03:52
qwen_imagen_server	03:54
qwen_edit_bundle	03:59
flux_2_dev	04:05
krea_2_turbo	04:05
text_to_audio	04:07
sfx_server	Aug 13 (same failure, undetected for 2 days)

Every one of them needed a manual restart. Nothing alerted — serverless health reported healthy, 0.0% error rate throughout, because it measures the HTTP invoke path, not the queue.

Why

The worker sat in blpop("ml_tasks") with no timeout.

That is a pure read-wait. The client transmits nothing, so the kernel never retransmits, never reaches tcp_retries2, and never raises. The thread parked forever on a socket the server had already discarded. This is why Redis reported blocked_clients: 0 while every worker believed it was blocked — the server had forgotten the connection; only the client still thought it existed.

Threads that were sending commands did surface the failure, as [Errno 110] Connection timed out about eighteen minutes later (the standard tcp_retries2 give-up window), and reconnected normally through _RedisWithRetry. The reader never could — _RedisWithRetry only retries on a raised error, and an unbounded BLPOP never raises.

The Redis server itself was healthy the whole time: 0 rejected connections, 0 evictions, 28 days uptime. The GPUs were idle at 29-31°C with the model still resident. This is a client bug.

The fix

1. blpop now takes BLPOP_TIMEOUT and the loop treats None as "keep waiting". Completing the read is what lets a dead socket be noticed at all. Kept below SOCKET_TIMEOUT because redis-py applies the socket read deadline to blocking commands too.
2. The connection pool sets socket_keepalive (with per-platform TCP knobs), socket_timeout, socket_connect_timeout, health_check_interval and retry_on_timeout, so a half-open connection gets reaped instead of trusted.
3. Background loops run their body inside _guarded_iteration. A bare while True: in a thread was one unhandled exception away from being gone permanently — the pruning thread died exactly that way on every replica of several endpoints while the process carried on looking fine.

Testing

6 new tests in tests/test_connection_liveness.py. Suite goes 51 → 57, no existing test touched.

Mutation-tested — each fix reverted individually, confirming the matching test turns red:

mutation	red test
restore unbounded blpop	test_worker_blpop_passes_a_timeout
strip pool keepalive/health-check	test_connection_pool_enables_half_open_detection
make _guarded_iteration re-raise	test_guarded_iteration_swallows_and_continues, test_heartbeat_loop_survives_a_raising_body

Notes for the reviewer

• Independent of perf: stop reading every task_result payload to prune keys Redis already expires #15. That PR removes the redundant prune_old_task_results SCAN storm, which is what triggered the pruning-thread crash; this PR stops any such error from being fatal. Both are worth having, in either order.
• **kwargs is deliberately not forwarded into ConnectionPool — callers pass unrelated kwargs to __init__ today and forwarding them would raise at connect time.
• Still open, not addressed here: self.worker_healthy is set True once at init and False at start_workers, and never back to True. A one-way latch — once tripped the worker refuses tasks until the process restarts. It did not cause this outage; it deserves its own PR and a deliberate re-check policy.
• Worth adding separately: a watchdog that exits non-zero when LLEN ml_tasks > 0 and nothing has been consumed for N minutes, so the orchestrator can act on a stall this class of bug cannot itself detect.

---

^{Need help on this PR? Tag @codesmith-bot with what you need. Autofix is enabled.}