Method reinitialize_shard_subscriptions

redis/asyncio/cluster.py:3711–3784

Reconcile per-node shard subscriptions against the cluster's current slot ownership map. For each tracked shard channel whose owning node has changed (e.g. after CLUSTER SETSLOT / failover), sunsubscribe on the old node's pubsub and ssubscribe on the new owner's pubs

(self)

                )

    async def reinitialize_shard_subscriptions(self) -> None:
        """
        Reconcile per-node shard subscriptions against the cluster's current
        slot ownership map. For each tracked shard channel whose owning node
        has changed (e.g. after CLUSTER SETSLOT / failover), sunsubscribe on
        the old node's pubsub and ssubscribe on the new owner's pubsub,
        preserving any registered handler.
        """
        uncovered: list = []
        made_progress = False
        first_migrate_error: Optional[BaseException] = None
        async with self._shard_state_lock:
            for channel, handler in list(self.shard_channels.items()):
                try:
                    new_node = self.cluster.get_node_from_key(channel)
                except SlotNotCoveredError:
                    # Slot is transiently uncovered (mid-migration / partial
                    # topology refresh). Defer this channel so coverable
                    # siblings still reconcile this pass; we surface the
                    # error below so the caller (and logs) know not every
                    # channel was reconciled. Retry happens on the next
                    # slots-cache change notification.
                    uncovered.append(channel)
                    continue
                old_name = self._shard_channel_to_node.get(channel)
                if old_name == new_node.name:
                    continue
                try:
                    await self._migrate_shard_channel(
                        channel, handler, old_name, new_node
                    )
                    made_progress = True
                except (ConnectionError, TimeoutError, OSError) as e:
                    # Transient connectivity error while subscribing on the
                    # new owner (or unsubscribing on the old owner if its
                    # handler chose to re-raise). Do not abort reconciliation
                    # for sibling channels: _shard_channel_to_node was not
                    # advanced for this channel, so the next slots-cache
                    # change notification will retry it.
                    logger.warning(
                        "shard channel %r migration deferred: %s: %s",
                        channel,
                        type(e).__name__,
                        e,
                    )
                    if first_migrate_error is None:
                        first_migrate_error = e
                    continue
            # Garbage-collect per-node pubsubs that no longer hold any
            # subscription so their connections are released.
            for name, pubsub in list(self.node_pubsub_mapping.items()):
                if not pubsub.subscribed:
                    try:
                        await pubsub.aclose()
                    except Exception:
                        pass
                    self.node_pubsub_mapping.pop(name, None)
        if uncovered: