[WA-10931] Make Redis queue polling use runtime-selectable RPOP

[xarus01]

Summary

• Redis transport now uses non-blocking RPOP for active queue polling by default.
• Kept the legacy BRPOP path as an emergency fallback via transport_options={'queue_pop_strategy': 'brpop'}.
• Redis Cluster transport is intentionally unchanged; with RPOP, the Envoy path should use the regular Redis transport.
• Empty RPOP scans are throttled with polling_interval; the BRPOP fallback disables that sleep/throttle to preserve the old blocking behavior.
• Updated focused Redis unit tests for the runtime switch, fallback path, event-loop behavior, and drain loop behavior.

Jira

• https://sendbird.atlassian.net/browse/WA-10931

Commits

• 3d0ebd1f - runtime Redis change (BRPOP to runtime-selectable RPOP, Redis-only).
• f0c3dba6 - Redis tests for the RPOP default, BRPOP fallback, and event-loop backlog polling.

Validation

uv run --python 3.11 --with pytest==8.3.5 --with amqp --with redis==5.2.1 pytest t/unit/transport/test_redis.py t/unit/transport/test_redis_cluster.py

Result: 137 passed.

git diff --check kombu-5.5.4-redis-cluster..HEAD

Result: clean.

[dittos]

RPOP re-poll cadence isn't driven by delivery

The throttle-skip-after-delivery in MultiChannelPoller.on_poll_start (redis.py:574-576) is the right fix for unconditional cooldown. But there's a related pacing concern worth thinking through before this lands: how does the hub actually decide when to call on_poll_start again after RPOP has just delivered a message? Looking at the hub loop helps frame it.

The hub loop (kombu/asynchronous/hub.py:301-380) is roughly:

while 1:
    poll_timeout = fire_timers(...) if scheduled else 1   # (a)
    for cb in self.on_tick: cb()                          # (b) on_poll_start runs here
    if readers or writers:
        events = poll(poll_timeout)                       # (c) uses (a)'s value
    else:
        sleep(min(poll_timeout, 0.1))

RPOP delivery doesn't signal the hub to come back sooner.
BRPOP gets its next-iteration re-arm "for free": its socket fd is registered, so a response wakes poll() at (c) and on_poll_start runs again immediately. RPOP registers no such fd. After on_poll_start delivers a message at (b), the hub still sleeps at (c) for whatever poll_timeout happened to be — driven by heartbeats, control fanout, scheduled timers, etc. The next RPOP attempt happens whenever something unrelated wakes the hub.

We correctly avoid arming the cooldown after a delivery, but the hub still goes to sleep at (c) before the next on_poll_start.

Concrete effects:

• Fill-rate is non-deterministic — two workers with identical prefetch_count/polling_interval can show different RPOP throughput depending on what else is registered on the hub.
• polling_interval only shortens the idle backoff — it doesn't control how fast we re-poll after a successful delivery.

Suggested fix: drain more within the same tick when RPOP returned a result.
The closest equivalent to BRPOP's "fd readiness re-arms immediately" is to keep pulling within on_poll_start while there's work and prefetch budget:

elif poll_queues:
    rpop_polled = True
    while channel.qos.can_consume():
        if not self._consume_from_active_queues(channel):
            break
        rpop_delivered = True

can_consume() provides a natural ceiling via prefetch_count. A per-tick cap may be worth adding so one busy channel doesn't starve other hub work, but the prefetch limit usually suffices.

Why not just schedule an immediate timer from on_poll_start?
That was my first instinct, but it doesn't work: poll_timeout is committed at (a), before (b). A call_later(0, ...) or call_soon(...) from inside the tick callback only affects the next iteration; the current one still sleeps the timeout it already committed to. There's no self-pipe in the hub for mid-poll interruption, so the in-tick drain above is the cleanest available approach.

Related: drain_events doesn't read polling_interval from transport_options.
Transport.drain_events (redis.py:1413-1417) uses self.polling_interval directly while register_with_event_loop honors transport_options['polling_interval']. Worth aligning these.