Batch SSUBSCRIBE by Redis Cluster slot to avoid CROSSSLOT errors and improve efficiency

[longcoding]

…ered bursts

Pull Request check-list

Please make sure to review and check all of these items:

• Does $ tox pass with this change (including linting)?
• 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?
• Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
• Is there an example added to the examples folder (if applicable)?
• Was the change added to CHANGES file?

NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open.

Description of change

This PR changes how SSUBSCRIBE commands are issued in Redis Cluster mode.

Previously, each channel was subscribed individually in a loop, resulting in one SSUBSCRIBE per channel. In Redis Cluster, this increases the risk of CROSSSLOT errors and introduces unnecessary overhead when multiple channels belong to the same hash slot.

This update improves the logic by grouping channels by their Redis hash slot, and sending a single SSUBSCRIBE per slot group, subscribing to all relevant channels at once.

Motivation

• Redis Cluster requires that all keys in a multi-key command reside in the same hash slot.
  Subscribing to multiple channels across slots in a single command causes CROSSSLOT errors.
• The previous implementation avoided this by sending individual commands per channel, but this is inefficient.
• Batching subscriptions by slot reduces command overhead and aligns with Redis Cluster's intended usage pattern.

Changes

• Group channels by their hash slot (key_slot(self.encoder.encode(channel)))
• Issue one SSUBSCRIBE command per slot group (i.e., SSUBSCRIBE channel1 channel2 ...)
• Add minor jitter between slot-level subscription requests to avoid synchronized load (±30%)

Example

channels_by_slot = defaultdict(list)
for channel in new_s_channels:
    slot = key_slot(self.encoder.encode(channel))
    channels_by_slot[slot].append(channel)

for slot, channels in channels_by_slot.items():
    self.execute_command("SSUBSCRIBE", *channels)
    time.sleep(base_interval * random.uniform(0.7, 1.3))

[bellatoris]

what about using 10. with current value if we use only one shard and slot count is close to like 16384 then we should wait 1600 sec to subscribe all existing channels...

[bellatoris]

closed by #28