redis/redis-py · error · ConnectionError

Failed to subscribe to cluster nodes

Error message

Failed to subscribe to cluster nodes: {', '.join(failed_nodes)}

What it means

In cluster keyspace-notification mode, refresh_subscriptions discovers current primaries and (p)subscribes each new/broken one. Any node whose psubscribe/subscribe throws is recorded; after all nodes are attempted a single ConnectionError is raised listing the failed node names. Successful nodes are kept, so the failure is partial. refresh_subscriptions runs automatically on topology change or connection errors during get_message().

Solutions

1. Retry: refresh_subscriptions re-runs automatically on the next get_message() error.
2. Check reachability and health of the named nodes.
3. Verify ACLs permit SUBSCRIBE/PSUBSCRIBE on every primary.
4. Let cluster topology stabilize (avoid mid-failover subscription storms).

from redis.exceptions import ConnectionError
for attempt in range(retries):
    try:
        return await pubsub.get_message(timeout=1)
    except ConnectionError as e:
        if 'Failed to subscribe to cluster nodes' not in str(e):
            raise
        await asyncio.sleep(backoff(attempt))
raise

When it happens

Trigger: A cluster failover or topology refresh where some primaries are unreachable or reject SUBSCRIBE/PSUBSCRIBE; triggered from get_message() error recovery or a manual refresh_subscriptions() call.

Common situations: Cluster rolling restart; transient network partition; ACLs on some nodes disallowing SUBSCRIBE; nodes temporarily down during a scale event.

Related errors

• Buffer is closed.
• Channel ' ' has no handler registered
• Connection has data
• Connection not ready
• Failed to subscribe to cluster nodes

AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/c5ca167f28ea36cd. Report an issue: GitHub.

Appendix: source

Thrown at redis/asyncio/keyspace_notifications.py:1019

            new_nodes = set(current_primaries.keys()) - set(self._node_pubsubs.keys())
            failed_nodes: list[str] = []
            for node_name in new_nodes:
                node = current_primaries[node_name]
                pubsub = await self._ensure_node_pubsub(node)

                try:
                    if self._subscribed_patterns:
                        await pubsub.psubscribe(**self._subscribed_patterns)
                    if self._subscribed_channels:
                        await pubsub.subscribe(**self._subscribed_channels)
                except Exception:
                    # Subscription failed - remove from dict so retry is possible
                    await self._cleanup_node(node_name)
                    failed_nodes.append(node_name)

            # Raise after attempting all nodes so we don't skip any
            if failed_nodes:
                raise ConnectionError(
                    f"Failed to subscribe to cluster nodes: {', '.join(failed_nodes)}"
                )

    async def aclose(self):
        """Close all pubsub connections and clean up resources."""
        self._closed = True
        for node_name in list(self._node_pubsubs.keys()):
            await self._cleanup_node(node_name)
        self._subscribed_patterns.clear()
        self._subscribed_channels.clear()

View on GitHub (pinned to 6a6b581b48)