this error can be related to a network issue, do you think it will be better to find a way to reinitialise the pubsub if that is the case.

Good point — you're right that a network error here could be transient and we shouldn't just surface it and give up.

I think the approach would be to catch network-level errors (connection refused, timeouts, EOF, etc.) during SSubscribe on the underlying PubSub, and when that happens:

1. Close the failed PubSub for that shard address
2. Remove it from the shards map so the next attempt gets a fresh connection
3. Re-resolve the node address for the affected channels (since the cluster state may have changed)
4. Retry the subscribe with a backoff, reusing the cluster's retryBackoff settings for consistency

This way transient network blips get retried transparently, and if the node is genuinely down the cluster state refresh in step 3 should route us to the new primary.

For the forwardMessages goroutine, I'd add similar logic — if the channel read returns an error/closes unexpectedly, attempt to re-establish that shard's PubSub and re-subscribe to the channels that were on it, rather than silently dropping messages.

Happy to implement this — I'll model the retry logic after how the existing PubSub.reconnect works internally so it stays consistent with the rest of the library.

Without any type of healthcheck here we cannot be sure if this pubsub connection is still opened.

That's a valid concern. Right now we create the PubSub via s.cluster.pubSub() and hand it off, but there's no ongoing verification that the connection is still alive.

I think the right approach is to add a Ping method on ShardedPubSub that iterates over all shard connections and pings each one. Something like:

func (s *ShardedPubSub) Ping(ctx context.Context) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    for addr, ps := range s.shards {
        if err := ps.Ping(ctx); err != nil {
            // connection is dead — trigger reconnect for this shard
            return fmt.Errorf("shard %s health check failed: %w", addr, err)
        }
    }
    return nil
}

On top of that, I could add an optional background health check loop (controlled by an option, disabled by default to avoid surprises) that periodically pings each shard and triggers reinitialization for any that have gone stale. This would pair well with the reconnect logic from the error handling comment — if a health check fails, we close the dead PubSub, re-resolve the node, and re-subscribe.

Makes sense to have at least the manual Ping method in this PR. Want me to add both the manual and the optional background health check, or keep the background one for a follow-up?

I think we can skip the health check for now and fail (change the underlying connection) if SSUBSCRIBE fails. In practice, making the "healtcheck" reactive than proactive.

makes sense -- went with the reactive approach. if SSubscribe fails, we close the broken shard, re-resolve the channels, and retry on a fresh connection. no background pinging.

Currently I don't see how this ShardedPubSub will react on topology changes (e.g. nodes are moved => shards are on a different address / new nodes are added => shards now split between nodes). Any plans or ideas how to address that with the current implementation?

Great question — this is the piece I knew would come up and it's definitely the trickiest part.

Currently the shards map is keyed by the node address resolved at subscribe time, and that mapping is completely static after that. If a slot migrates to a different node, we'd keep reading from the old node (which would stop receiving messages for those channels) and never know about it.

Here's my plan for handling this:

1. Hook into cluster state refresh
The ClusterClient already periodically reloads cluster state via state.Get(). After each topology refresh, ShardedPubSub should re-resolve every channel's node address and compare against the current mapping. For any channel whose node has changed:

• Subscribe to the channel on the new node's PubSub
• Unsubscribe from the old node's PubSub
• If an old node's PubSub has no remaining channels, close it and remove it from the map

2. Listen for MOVED responses
If any operation returns a MOVED error, that's an immediate signal that topology has shifted. This could trigger an on-demand re-resolve for the affected shard rather than waiting for the next periodic refresh.

3. Notification channel for resharding events
This might be useful for consumers who want to know when migrations happened, but that feels like it could be a follow-up.

I think items 1 and 2 are important to get right in this PR. The implementation would look something like adding a refreshTopology(ctx) method that gets called either periodically (piggy-backing on the cluster's reload interval) or reactively on errors. I'll work on this — would appreciate your input on whether you'd prefer it as a periodic goroutine or event-driven off the cluster state reload.

@veeceey I do think that we should have the ClusterClient as the source of truth for the topology and that ShardedPubSub should update (or just recalculate, if an update is not needed) its map when ClusterClient updates its topology. Hence, something like an event driven approach would fit best here.

Since the ShardedPubSub is a specific tool used only in the ClusterClient feel free to have a set (slice) of all ShardedPubSubs somehowe registered on the ClusterClient and trigger their topology changes when the ClusterClient updates its topology. It would be way simpler, somehowe limiting (to just covering this case, instead of generic event channel), but a good balance in my mind.

Let me know if you find any problems with this approach or thing another one fits better, I am willing to review alternatives as well.

that makes a lot of sense — having the ClusterClient own the topology and push updates to registered ShardedPubSubs is way cleaner than each pubsub polling independently. I like the simplicity of a slice of registered instances on the ClusterClient.

I'm thinking something like:

• add a shardedPubSubs []*ShardedPubSub field on ClusterClient (protected by the existing mutex)
• ShardedPubSub registers itself on creation, deregisters on Close
• when ClusterClient refreshes topology (in reloadState or wherever that happens), it calls a onTopologyChange() method on each registered ShardedPubSub
• that method re-resolves channels to slots/nodes and migrates subscriptions as needed

I don't see any issues with this approach — it keeps things scoped to exactly this use case without over-engineering a generic event system. I'll start working on it and push an update. appreciate the guidance!

a slight optimization with a map (or even sync.Map) can be done here, but I think this is fine for now. I don't think we can expect such number of sharded pubsubs that looping over them may be a problem.

resubscribeShard is called while holding s.mu, but internally it calls nodeAddrForChannel, which calls slotMasterNode, which calls state.Get.

• If the cluster state is nil, Get calls Reload which will eventually call notifyShardedPubSubs, which calls onTopologyChange trying to acquire s.mu.

Release the lock before the retry, similar to how the main SSubscribe flow already does it for address resolution.

good catch — you're right, resubscribeShard holding s.mu while calling nodeAddrForChannel creates the same deadlock path as the original SSubscribe issue. I'll release the lock before the retry and re-acquire after, same pattern as the main SSubscribe flow.

since we will hold the lock for applying the migration, does it make sense to verify the channels still exist when subscribing?

    if _, exists := s.chanShard[ch]; exists {
        s.chanShard[ch] = newAddr
    }

yep, makes total sense — if someone unsubscribed while we were resolving, we shouldn't silently re-add it. will add the existence check before updating chanShard.

I think we can deregister before acquiring the s.mu.Lock()

agreed — deregistering doesn't need the lock and doing it first avoids holding s.mu longer than necessary. will move it up.

Can this be async? maybe with validating that it is not triggered multiple times. If you look at LazyReload it has a mechanism for executing the latest reload, maybe something similar here would help.

yeah, making it async would avoid blocking the reload path. I'll look at LazyReload's dedup mechanism and model it similarly — signal a goroutine to do the migration instead of doing it inline, with a check to skip if one's already in-flight.

The existing PubSub supports Ping, ReceiveMessage and ChannelWithSubscriptions. ShardedPubSub only exposes Channel(). Can you either add them for feature parity or document that they are intentionally not supported? I would like to have a PubSub interface in the future (for v10 for example), where we can return the same Interface for the pubsub and sharded pubsub.

that's a fair point for v10 interface parity. I'll add Ping and ReceiveMessage methods on ShardedPubSub — Ping iterates over shards and pings each, ReceiveMessage reads from the merged channel. ChannelWithSubscriptions I can stub for now with a TODO since the subscription tracking is a bit different with sharded channels. Will document what's supported vs not.

While theoretically we should connect to the replica on read only, I am wondering if we want to? The master will receive the message first. I will validate if this is needed at all or should be changed, but for now let's keep it as it is.

sounds good, I'll leave it as-is for now then. makes sense to validate whether replica reads are actually useful for pubsub before changing it.

If SSubscribe fails (error is swallowed with _), the channel is now:

• Unsubscribed from the old shard
• NOT subscribed on the new shard
• BUT chanShard is updated to point to the new shard
  The channel is effectively lost - no messages will be received, and the code thinks everything is fine.

Can we update the mapping ONLY on successful SSubscribe here?

you're right, that's a nasty silent failure. I'll restructure so we only update chanShard[ch] = newAddr after SSubscribe succeeds. if it fails, we keep the old mapping so at least the channel isn't orphaned — and we can log or return the error so it's visible.