fix: honor read_from_replicas for sharded pubsub subscriptions in ClusterPubSub

[Brumbelow]

Description of change

Fixes #3266

ClusterPubSub routes every sharded subscription to the slot's primary: ssubscribe, the sunsubscribe fallback, and reinitialize_shard_subscriptions resolve nodes via get_node_from_key(), which ignores read_from_replicas and load_balancing_strategy — unlike regular pub/sub node selection in ClusterPubSub.execute_command, which honors both.

Shard-channel resolution now goes through the same nodes_manager.get_node_from_slot(slot, read_from_replicas, load_balancing_strategy) in the sync and async clients. Since a shard subscription may now live on a replica, the keep-vs-migrate decisions (slot-migration reconciliation and the lazy re-route in ssubscribe) check whether the tracked node still serves the channel's slot instead of comparing against a fresh pick, so subscriptions don't bounce between shard siblings on topology refreshes.

With the default configuration (read_from_replicas=False, no load balancing strategy) behavior is unchanged: subscriptions resolve to and strictly follow the slot primary, and SlotNotCoveredError is still raised for uncovered slots.

Pull Request check-list

• Do tests and lints pass with this change?
• Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
• Is the new or changed code fully tested?
• [N/A] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
• [N/A] Is there an example added to the examples folder (if applicable)?

---

Note

Medium Risk
Touches cluster pub/sub routing and migration/reconcile paths in both sync and async clients; behavior changes only when replica reads or load balancing is enabled, but mistakes could drop or duplicate shard subscriptions during failover.

Overview
Fixes #3266 by aligning shard pub/sub node selection in sync and async ClusterPubSub with regular cluster routing: ssubscribe, sunsubscribe fallback, and reinitialize_shard_subscriptions now use nodes_manager.get_node_from_slot(slot, read_from_replicas, load_balancing_strategy) instead of get_node_from_key() (primary-only).

A new _eligible_subscription_nodes helper defines which nodes may hold a subscription for a slot (replicas included only when replica reads or load balancing is enabled). Keep-vs-migrate logic no longer compares the tracked node to a fresh balancer pick; it skips migration when the tracked node still serves the slot, avoiding churn between primary and replica on re-subscribe or topology refresh. sunsubscribe without a reverse-index entry scans eligible slot nodes for the pubsub that actually holds the channel.

Default config (read_from_replicas=False) behavior is unchanged. Tests are updated for the new resolution path and extended for replica routing, sibling stability, and unsubscribe fallback.

^{Reviewed by Cursor Bugbot for commit c2f60ee. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit c2f60ee. Configure here.}

[petyaslavova]

Hey @Brumbelow, thank you for your contribution!

You're right that the server side allows this — Redis forwards shard messages to all nodes in the shard, so a client may subscribe to a shard channel on the slot's primary or on any of its replicas, and we already send READONLY on node connections whenever read_from_replicas/load_balancing_strategy is set.

Before we take this toward merge there is a design question to settle on our side: those two options are documented as read-command routing knobs (and read_from_replicas is deprecated), so folding shard-subscription placement into them silently changes the delivery path for existing users — a replica that is reachable by the client but partitioned from its primary stops receiving shard messages with nothing to notice it. We'd also want placement sticky per shard rather than round-robin per channel, so channels in one slot don't spread across siblings and enlarge the polled pubsub set.

Two points are blockers regardless of that decision:

• Async NodesManager.get_node_from_slot only converts IndexError/TypeError, so an uncovered slot now surfaces as a bare KeyError. The except SlotNotCoveredError deferral in reinitialize_shard_subscriptions no longer fires and a whole reconcile pass aborts, and ssubscribe no longer matches the sync error type. The updated unit test hides this because the mock raises SlotNotCoveredError itself. Bugbot flagged this one as well.
• The new coverage is mock-only, so nothing proves the actual behavior this PR relies on. Our CI cluster runs with one replica per shard and we already have onlycluster shard-pubsub tests — please add an end-to-end test that subscribes with replica routing enabled and receives an SPUBLISH published to the shard.

Please keep this as a draft in the meantime. I'll follow up on #3266 for the operational scenario behind the request, which is what decides whether we do this at all and whether it should be an explicit opt-in.

[Brumbelow]

Thanks @petyaslavova , both blockers are pushed and addressed.

• Async NodesManager.get_node_from_slot now raises SlotNotCoveredError when the slot is missing from slots_cache, matching the sync nodes manager, so the deferral in reinitialize_shard_subscriptions fires again and ssubscribe raises the same error type as sync.
• Added end-to-end coverage (TestClusterPubSubReplicaRouting, sync and async): with read_from_replicas=True, two channels in the same slot are subscribed so one lands on the shard's replica, then both receive a PUBLISH

Will keep as draft until you all tell me otherwise. Let me know if I can help!