fix: keep Redis SUBSCRIBE alive by polling get_message instead of listen()

[tjluyao]

Root cause (chronic fleet dispatch death)

Field nodes go dispatch-dead ~1 h after every (re)start: registered + heartbeating, but PUBSUB NUMSUB node:<id>:dispatch = 0, so the host logs No subscribers on tasks channel and tasks fail max_attempts_exceeded. Restarting a box fixes it for ~1 h, then it recurs. A prior change added socket_keepalive + socket_keepalive_options (TCP_KEEPIDLE 60/…) + health_check_interval=30 on the client, but the deaths continued.

The reason: all three listeners (dispatch, cmds, node responses) consume through iter_pubsub_messages, which looped over pubsub.listen(). listen() parks inside a single parse_response(block=True) for the entire idle lifetime of the subscription — redis-py only runs check_health() on entry to parse_response, so on a quiet channel the health-check PING fires at most once and then never again. With no application-level traffic, the LB's idle timeout_server culls the SUBSCRIBE connection (TCP keepalive ACKs are L4 and don't reset HAProxy's L7 inactivity timer). health_check_interval was set but never driven.

Fix

Consume with get_message(timeout=1.0) in a loop instead of listen(). Every iteration calls check_health(), which emits the periodic PING (every health_check_interval=30 s) that keeps the connection warm and the subscription registered — well under any idle timeout. One-line-of-behavior change in the shared helper, so it fixes the dispatch listener, command listener, and node-response listener at once.

Tests

• New tests/server/test_iter_pubsub_messages.py: asserts it polls via get_message(timeout) (never listen()), idle None ticks don't end iteration, bad-JSON is skipped, and ConnectionError stops cleanly.
• Full server suite: 435 passed.

Deploy

Server-only change → new flowmesh_server image, rolled to the fleet (same recipe as the rc.3 rollout). Complements #95 (dispatcher no longer mis-reports undeliverable dispatch) and deploy_infra #8 (watchdog force-restart as a safety net). This PR removes the need for either as a recovery mechanism by preventing the cull in the first place.