fix(ssubscribe): properly resubscribe in case of shard failover (redis#3098)

* fix(ssubscribe): properly resubscribe in case of shard failover

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: Cluster resubscribes all listeners of the failed client ( possibly some/all of those will end up in a different client after the rediscovery ) 

fixes: redis#2902

Author: nkaradzhov

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

@@ -338,6 +338,10 @@ export default class RedisCommandsQueue {
     return this.#addPubSubCommand(command);
   }
 
+  removeAllPubSubListeners() {
+    return this.#pubSub.removeAllListeners();
+  }
+
   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', this._self.#queue.removeAllPubSubListeners());
+        }
+      });
       if (resubscribePromise) {
         promises.push(resubscribePromise);
       }

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,22 @@ export class PubSub {
     return listeners;
   }
 
+  removeAllListeners() {
+    const result = {
+      [PUBSUB_TYPE.CHANNELS]: this.listeners[PUBSUB_TYPE.CHANNELS],
+      [PUBSUB_TYPE.PATTERNS]: this.listeners[PUBSUB_TYPE.PATTERNS],
+      [PUBSUB_TYPE.SHARDED]: this.listeners[PUBSUB_TYPE.SHARDED]
+    }
+
+    this.#updateIsActive();
+
+    this.listeners[PUBSUB_TYPE.CHANNELS] = new Map();
+    this.listeners[PUBSUB_TYPE.PATTERNS] = new Map();
+    this.listeners[PUBSUB_TYPE.SHARDED] = new Map();
+
+    return result;
+  }
+
   #emitPubSubMessage(
     type: PubSubType,
     message: Buffer,

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

@@ -2,7 +2,7 @@ import { RedisClusterClientOptions, RedisClusterOptions } from '.';
 import { RootNodesUnavailableError } from '../errors';
 import RedisClient, { RedisClientOptions, RedisClientType } from '../client';
 import { EventEmitter } from 'node:stream';
-import { ChannelListeners, PUBSUB_TYPE, PubSubTypeListeners } from '../client/pub-sub';
+import { ChannelListeners, PUBSUB_TYPE, PubSubListeners, PubSubTypeListeners } from '../client/pub-sub';
 import { RedisArgument, RedisFunctions, RedisModules, RedisScripts, RespVersions, TypeMapping } from '../RESP/types';
 import calculateSlot from 'cluster-key-slot';
 import { RedisSocketOptions } from '../client/socket';
@@ -185,6 +185,7 @@ export default class RedisClusterSlots<
   async #discover(rootNode: RedisClusterClientOptions) {
     this.clientSideCache?.clear();
     this.clientSideCache?.disable();
+
     try {
       const addressesInUse = new Set<string>(),
         promises: Array<Promise<unknown>> = [],
@@ -224,6 +225,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;
 
@@ -337,23 +339,29 @@ export default class RedisClusterSlots<
     const socket =
       this.#getNodeAddress(node.address) ??
       { host: node.host, port: node.port, };
-    const client = Object.freeze({
+    const clientInfo = Object.freeze({
       host: socket.host,
       port: socket.port,
     });
     const emit = this.#emit;
-    return this.#clientFactory(
+    const client = this.#clientFactory(
       this.#clientOptionsDefaults({
         clientSideCache: this.clientSideCache,
         RESP: this.#options.RESP,
         socket,
         readonly,
       }))
-      .on('error', error => emit('node-error', error, client))
-      .on('reconnecting', () => emit('node-reconnecting', client))
-      .once('ready', () => emit('node-ready', client))
-      .once('connect', () => emit('node-connect', client))
-      .once('end', () => emit('node-disconnect', client));
+      .on('error', error => emit('node-error', error, clientInfo))
+      .on('reconnecting', () => emit('node-reconnecting', clientInfo))
+      .once('ready', () => emit('node-ready', clientInfo))
+      .once('connect', () => emit('node-connect', clientInfo))
+      .once('end', () => emit('node-disconnect', clientInfo))
+      .on('__MOVED', async (allPubSubListeners: PubSubListeners) => {
+        await this.rediscover(client);
+        this.#emit('__resubscribeAllPubSubListeners', allPubSubListeners);
+      });
+
+    return client;
   }
 
   #createNodeClient(node: ShardNode<M, F, S, RESP, TYPE_MAPPING>, readonly?: boolean) {
@@ -374,7 +382,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 { PubSubListener, PubSubListeners } 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('__resubscribeAllPubSubListeners', this.resubscribeAllPubSubListeners.bind(this));
 
     if (options?.commandOptions) {
       this._commandOptions = options.commandOptions;
@@ -584,6 +585,33 @@ export default class RedisCluster<
     );
   }
 
+  resubscribeAllPubSubListeners(allListeners: PubSubListeners) {
+    for(const [channel, listeners] of allListeners.CHANNELS) {
+      listeners.buffers.forEach(bufListener => {
+        this.subscribe(channel, bufListener, true);
+      });
+      listeners.strings.forEach(strListener => {
+        this.subscribe(channel, strListener);
+      });
+    };
+    for (const [channel, listeners] of allListeners.PATTERNS) {
+      listeners.buffers.forEach(bufListener => {
+        this.pSubscribe(channel, bufListener, true);
+      });
+      listeners.strings.forEach(strListener => {
+        this.pSubscribe(channel, strListener);
+      });
+    };
+    for (const [channel, listeners] of allListeners.SHARDED) {
+      listeners.buffers.forEach(bufListener => {
+        this.sSubscribe(channel, bufListener, true);
+      });
+      listeners.strings.forEach(strListener => {
+        this.sSubscribe(channel, strListener);
+      });
+    };
+  }
+
   sUnsubscribe = this.SUNSUBSCRIBE;
 
   /**