fix(ssubscribe): properly resubscribe in case of shard failover in RE TODO cleanup debug logs

1) when RE failover happens, there is a disconnect
2) affected Client reconnects and tries to resubscribe all existing listeners
ISSUE #1: CROSSSLOT Error - client was doing ssubscribe ch1 ch2.. chN which, after the failover could result in CROSSSLOT ( naturally, becasuse now some slots could be owned by other shards )
FIX: send one ssubscribe command per channel instead of one ssubscribe for all channels
ISSUE #2: MOVED Error - some/all of the channels might be moved somewhere else
FIX: 1: propagate the error to the Cluster. 2: Cluster rediscovers topology.
3: Extract all existing subscriptions from all pubsub clients and resubscribe
over the new topology.

fixes: #2902

Author: nkaradzhov

packages/client/lib/client/commands-queue.ts

@@ -320,6 +320,7 @@ export default class RedisCommandsQueue {
     listener?: PubSubListener<T>,
     returnBuffers?: T
   ) {
+    console.log(`CQ::unsubscribe(${channels})`);
     const command = this.#pubSub.unsubscribe(type, channels, listener, returnBuffers);
     if (!command) return;
 
@@ -338,6 +339,14 @@ export default class RedisCommandsQueue {
     return this.#addPubSubCommand(command);
   }
 
+  getShardedChannels(): IterableIterator<string> {
+    return this.#pubSub.getShardedChannels();
+  }
+
+  removeShardedListeners(channel: string): ChannelListeners {
+    return this.#pubSub.removeShardedListeners(channel);
+  }
+
   resubscribe(chainId?: symbol) {
     const commands = this.#pubSub.resubscribe();
     if (!commands.length) return;

packages/client/lib/client/index.ts

@@ -765,17 +765,17 @@ export default class RedisClient<
         }
       });
     }
-    
+
     if (this.#clientSideCache) {
       commands.push({cmd: this.#clientSideCache.trackingOn()});
     }
 
     if (this.#options?.emitInvalidate) {
       commands.push({cmd: ['CLIENT', 'TRACKING', 'ON']});
     }
-    
+
     const maintenanceHandshakeCmd = await EnterpriseMaintenanceManager.getHandshakeCommand(this.#options);
-    
+
     if(maintenanceHandshakeCmd) {
       commands.push(maintenanceHandshakeCmd);
     };
@@ -818,6 +818,11 @@ export default class RedisClient<
         chainId = Symbol('Socket Initiator');
 
       const resubscribePromise = this.#queue.resubscribe(chainId);
+      resubscribePromise?.catch(error => {
+        if (error.message && error.message.startsWith('MOVED')) {
+          this.emit('__MOVED')
+        }
+      });
       if (resubscribePromise) {
         promises.push(resubscribePromise);
       }
@@ -1192,6 +1197,14 @@ export default class RedisClient<
 
   sUnsubscribe = this.SUNSUBSCRIBE;
 
+  getShardedChannels(): IterableIterator<string> {
+    return this._self.#queue.getShardedChannels();
+  }
+
+  removeShardedListeners(channel: string): ChannelListeners {
+    return this._self.#queue.removeShardedListeners(channel);
+  }
+
   async WATCH(key: RedisVariadicArgument) {
     const reply = await this._self.sendCommand(
       pushVariadicArguments(['WATCH'], key)

packages/client/lib/client/pub-sub.ts

@@ -323,25 +323,50 @@ export class PubSub {
   }
 
   resubscribe() {
-    const commands = [];
+    const commands: PubSubCommand[] = [];
     for (const [type, listeners] of Object.entries(this.listeners)) {
       if (!listeners.size) continue;
 
       this.#isActive = true;
+
+      if(type === PUBSUB_TYPE.SHARDED) {
+        this.#shardedResubscribe(commands, listeners);
+      } else {
+        this.#normalResubscribe(commands, type, listeners);
+      }
+    }
+
+    return commands;
+  }
+
+  #normalResubscribe(commands: PubSubCommand[], type: string, listeners: PubSubTypeListeners) {
+    this.#subscribing++;
+    const callback = () => this.#subscribing--;
+    commands.push({
+      args: [
+        COMMANDS[type as PubSubType].subscribe,
+        ...listeners.keys()
+      ],
+      channelsCounter: listeners.size,
+      resolve: callback,
+      reject: callback
+    });
+  }
+
+  #shardedResubscribe(commands: PubSubCommand[], listeners: PubSubTypeListeners) {
+    const callback = () => this.#subscribing--;
+    for(const channel of listeners.keys()) {
       this.#subscribing++;
-      const callback = () => this.#subscribing--;
       commands.push({
         args: [
-          COMMANDS[type as PubSubType].subscribe,
-          ...listeners.keys()
+          COMMANDS[PUBSUB_TYPE.SHARDED].subscribe,
+          channel
         ],
-        channelsCounter: listeners.size,
+        channelsCounter: 1,
         resolve: callback,
         reject: callback
-      } satisfies PubSubCommand);
+      })
     }
-
-    return commands;
   }
 
   handleMessageReply(reply: Array<Buffer>): boolean {
@@ -379,6 +404,10 @@ export class PubSub {
     return listeners;
   }
 
+  getShardedChannels(): IterableIterator<string>  {
+    return this.listeners[PUBSUB_TYPE.SHARDED].keys()
+  }
+
   #emitPubSubMessage(
     type: PubSubType,
     message: Buffer,

packages/client/lib/cluster/cluster-slots.ts

@@ -175,6 +175,7 @@ export default class RedisClusterSlots<
     throw new RootNodesUnavailableError();
   }
 
+  //TODO investigate what happens with the masters and replicas. It looks like they
   #resetSlots() {
     this.slots = new Array(RedisClusterSlots.#SLOTS);
     this.masters = [];
@@ -185,6 +186,22 @@ export default class RedisClusterSlots<
   async #discover(rootNode: RedisClusterClientOptions) {
     this.clientSideCache?.clear();
     this.clientSideCache?.disable();
+
+
+    const allChannelListeners = new Map<string, ChannelListeners>();
+
+    for (const master of this.masters) {
+      const shardedClient = master.pubSub?.client;
+      if (!shardedClient) continue;
+      for (const channel of shardedClient.getShardedChannels()) {
+        const listeners = shardedClient.removeShardedListeners(channel);
+        if(allChannelListeners.get(channel)) {
+          console.warn(`Found existing listeners, will be overwritten...`);
+        }
+        allChannelListeners.set(channel, listeners);
+      }
+    }
+
     try {
       const addressesInUse = new Set<string>(),
         promises: Array<Promise<unknown>> = [],
@@ -224,6 +241,7 @@ export default class RedisClusterSlots<
         }
       }
 
+      //Keep only the nodes that are still in use
       for (const [address, node] of this.nodeByAddress.entries()) {
         if (addressesInUse.has(address)) continue;
 
@@ -239,6 +257,9 @@ export default class RedisClusterSlots<
         this.nodeByAddress.delete(address);
       }
 
+      this.#emit('__refreshShardedChannels', allChannelListeners);
+
+
       await Promise.all(promises);
       this.clientSideCache?.enable();
 
@@ -354,6 +375,9 @@ export default class RedisClusterSlots<
       .once('ready', () => emit('node-ready', client))
       .once('connect', () => emit('node-connect', client))
       .once('end', () => emit('node-disconnect', client));
+      .on('__MOVED', () => {
+        this.rediscover(client);
+      })
   }
 
   #createNodeClient(node: ShardNode<M, F, S, RESP, TYPE_MAPPING>, readonly?: boolean) {
@@ -374,7 +398,9 @@ export default class RedisClusterSlots<
 
   async rediscover(startWith: RedisClientType<M, F, S, RESP>): Promise<void> {
     this.#runningRediscoverPromise ??= this.#rediscover(startWith)
-      .finally(() => this.#runningRediscoverPromise = undefined);
+      .finally(() => {
+        this.#runningRediscoverPromise = undefined
+      });
     return this.#runningRediscoverPromise;
   }
 

packages/client/lib/cluster/index.ts

@@ -6,7 +6,7 @@ import { EventEmitter } from 'node:events';
 import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumentsPrefix } from '../commander';
 import RedisClusterSlots, { NodeAddressMap, ShardNode } from './cluster-slots';
 import RedisClusterMultiCommand, { RedisClusterMultiCommandType } from './multi-command';
-import { PubSubListener } from '../client/pub-sub';
+import { ChannelListeners, PubSubListener } from '../client/pub-sub';
 import { ErrorReply } from '../errors';
 import { RedisTcpSocketOptions } from '../client/socket';
 import { ClientSideCacheConfig, PooledClientSideCacheProvider } from '../client/cache';
@@ -310,6 +310,7 @@ export default class RedisCluster<
 
     this._options = options;
     this._slots = new RedisClusterSlots(options, this.emit.bind(this));
+    this.on('__refreshShardedChannels', this.refreshShardedChannelsSubscriptions.bind(this));
 
     if (options?.commandOptions) {
       this._commandOptions = options.commandOptions;
@@ -449,6 +450,7 @@ export default class RedisCluster<
         }
 
         if (err.message.startsWith('MOVED')) {
+          console.log('Cluster::_execute() -> MOVED, rediscover');
           await this._slots.rediscover(client);
           client = await this._slots.getClient(firstKey, isReadonly);
           continue;
@@ -553,6 +555,7 @@ export default class RedisCluster<
       firstChannel = Array.isArray(channels) ? channels[0] : channels;
     let client = await this._self._slots.getShardedPubSubClient(firstChannel);
     for (let i = 0; ; i++) {
+      console.log(`SSUBSCRIBE ch:${channels} try(${i}`);
       try {
         return await client.SSUBSCRIBE(channels, listener, bufferMode);
       } catch (err) {
@@ -561,6 +564,7 @@ export default class RedisCluster<
         }
 
         if (err.message.startsWith('MOVED')) {
+          console.log(`RedisCluster::SSUBSCRIBE() -> MOVED, rediscover`);
           await this._self._slots.rediscover(client);
           client = await this._self._slots.getShardedPubSubClient(firstChannel);
           continue;
@@ -584,6 +588,17 @@ export default class RedisCluster<
     );
   }
 
+  refreshShardedChannelsSubscriptions(allChannelListeners: Map<string, ChannelListeners>) {
+    for(const [channel, listeners] of allChannelListeners) {
+      for(const bufListener of listeners.buffers) {
+        this.sSubscribe(channel, bufListener, true);
+      }
+      for(const strListener of listeners.strings) {
+        this.sSubscribe(channel, strListener);
+      }
+    }
+  }
+
   sUnsubscribe = this.SUNSUBSCRIBE;
 
   /**