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/client-counter.ts

@@ -0,0 +1,19 @@
+class ClientCounter {
+  #id = 0;
+  #activeClients: number[] = [];
+
+  getNextId(): number {
+    this.#activeClients.push(this.#id);
+    return this.#id++;
+  }
+
+  getActiveClients() {
+    return { total: this.#activeClients.length, clients: this.#activeClients };
+  }
+
+  removeClient(id: number) {
+    this.#activeClients = this.#activeClients.filter((cid) => cid !== id);
+  }
+}
+
+export default new ClientCounter();

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

@@ -21,6 +21,7 @@ import { BasicCommandParser, CommandParser } from './parser';
 import SingleEntryCache from '../single-entry-cache';
 import { version } from '../../package.json'
 import EnterpriseMaintenanceManager, { MaintenanceUpdate, MovingEndpointType } from './enterprise-maintenance-manager';
+import clientCounter from './client-counter';
 
 export interface RedisClientOptions<
   M extends RedisModules = RedisModules,
@@ -500,8 +501,10 @@ export default class RedisClient<
     this._self.#dirtyWatch = msg;
   }
 
+  id: number
   constructor(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>) {
     super();
+    this.id = clientCounter.getNextId();
     this.#validateOptions(options)
     this.#options = this.#initiateOptions(options);
     this.#queue = this.#initiateQueue();
@@ -765,17 +768,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);
     };
@@ -812,12 +815,22 @@ export default class RedisClient<
     .on('end', () => this.emit('end'));
   }
 
+  #getHost() {
+    //@ts-ignore
+    return this.options?.url ?? this.options?.socket?.host
+  }
+
   #initiateSocket(): RedisSocket {
     const socketInitiator = async () => {
       const promises = [],
         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);
       }
@@ -953,6 +966,7 @@ export default class RedisClient<
   }
 
   async connect() {
+    console.log(`Create Client${this._self.id} ${this._self.#getHost()}`);
     await this._self.#socket.connect();
     return this as unknown as RedisClientType<M, F, S, RESP, TYPE_MAPPING>;
   }
@@ -1192,6 +1206,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)
@@ -1544,6 +1566,9 @@ export default class RedisClient<
    * Destroy the client. Rejects all commands immediately.
    */
   destroy() {
+    //@ts-ignore
+    console.log(`Destroy Client(${this.id}) ${this._self.#getHost()}`);
+    clientCounter.removeClient(this.id);
     clearTimeout(this._self.#pingTimer);
     this._self.#queue.flushAll(new DisconnectsClientError());
     this._self.#socket.destroy();

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

@@ -7,6 +7,7 @@ import { RedisArgument, RedisFunctions, RedisModules, RedisScripts, RespVersions
 import calculateSlot from 'cluster-key-slot';
 import { RedisSocketOptions } from '../client/socket';
 import { BasicPooledClientSideCache, PooledClientSideCacheProvider } from '../client/cache';
+import clientCounter from '../client/client-counter';
 
 interface NodeAddress {
   host: string;
@@ -161,6 +162,7 @@ export default class RedisClusterSlots<
   }
 
   async #discoverWithRootNodes() {
+    console.log('CS::#discoverWithRootNodes()');
     let start = Math.floor(Math.random() * this.#options.rootNodes.length);
     for (let i = start; i < this.#options.rootNodes.length; i++) {
       if (!this.#isOpen) throw new Error('Cluster closed');
@@ -175,6 +177,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 = [];
@@ -183,16 +186,41 @@ export default class RedisClusterSlots<
   }
 
   async #discover(rootNode: RedisClusterClientOptions) {
+    console.log('CS::discover()', rootNode.socket?.host);
+    console.log(`Clients before`, clientCounter.getActiveClients());
     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);
+        const existingListeners = allChannelListeners.get(channel);
+        if(existingListeners) {
+          console.log(`hmmm, why are there existing listeners?`, existingListeners, existingListeners === listeners);
+        } else {
+          allChannelListeners.set(channel, listeners);
+        }
+      }
+    }
+
+    console.log(`we have extracted all listeners`, allChannelListeners);
+
     try {
       const addressesInUse = new Set<string>(),
         promises: Array<Promise<unknown>> = [],
         eagerConnect = this.#options.minimizeConnections !== true;
 
       const shards = await this.#getShards(rootNode);
+      console.log('Shards', shards.map(s => `${s.from}-${s.to} -> ${s.master.host}`));
       this.#resetSlots(); // Reset slots AFTER shards have been fetched to prevent a race condition
       for (const { from, to, master, replicas } of shards) {
+        // TODO check if old slot nodes are disconnected and deleted
+        console.log(`Creating shard?`);
         const shard: Shard<M, F, S, RESP, TYPE_MAPPING> = {
           master: this.#initiateSlotNode(master, false, eagerConnect, addressesInUse, promises)
         };
@@ -209,6 +237,7 @@ export default class RedisClusterSlots<
       }
 
       if (this.pubSubNode && !addressesInUse.has(this.pubSubNode.address)) {
+        console.log('refresh pubsubnode?');
         const channelsListeners = this.pubSubNode.client.getPubSubListeners(PUBSUB_TYPE.CHANNELS),
           patternsListeners = this.pubSubNode.client.getPubSubListeners(PUBSUB_TYPE.PATTERNS);
 
@@ -224,7 +253,12 @@ export default class RedisClusterSlots<
         }
       }
 
+      console.log('addressesInUse', addressesInUse);
+      console.log('nodeByAddress', this.nodeByAddress.keys());
+      //Keep only the nodes that are still in use
       for (const [address, node] of this.nodeByAddress.entries()) {
+        //@ts-ignore
+        console.log(address, 'client:', !!node.client, 'pubsub client:', !!node.pubSub);
         if (addressesInUse.has(address)) continue;
 
         if (node.client) {
@@ -239,11 +273,16 @@ export default class RedisClusterSlots<
         this.nodeByAddress.delete(address);
       }
 
+      this.#emit('__refreshShardedChannels', allChannelListeners);
+
+
       await Promise.all(promises);
       this.clientSideCache?.enable();
 
+      console.log(`Clients after`, clientCounter.getActiveClients());
       return true;
     } catch (err) {
+      console.log('ERROR REDISCOVERING', err);
       this.#emit('error', err);
       return false;
     }
@@ -257,6 +296,7 @@ export default class RedisClusterSlots<
     options.commandOptions = undefined;
 
     // TODO: find a way to avoid type casting
+    console.log('create client to get shards');
     const client = await this.#clientFactory(options as RedisClientOptions<M, F, S, RESP, {}>)
       .on('error', err => this.#emit('error', err))
       .connect();
@@ -354,9 +394,14 @@ export default class RedisClusterSlots<
       .once('ready', () => emit('node-ready', client))
       .once('connect', () => emit('node-connect', client))
       .once('end', () => emit('node-disconnect', client));
+      .on('__MOVED', () => {
+        console.log(`cluster::createClient -> MOVED, rediscover`);
+        this.rediscover(client);
+      })
   }
 
   #createNodeClient(node: ShardNode<M, F, S, RESP, TYPE_MAPPING>, readonly?: boolean) {
+    console.log("CS::#createNodeClient", node.host);
     const client = node.client = this.#createClient(node, readonly);
     return node.connectPromise = client.connect()
       .finally(() => node.connectPromise = undefined);
@@ -373,8 +418,12 @@ export default class RedisClusterSlots<
   #runningRediscoverPromise?: Promise<void>;
 
   async rediscover(startWith: RedisClientType<M, F, S, RESP>): Promise<void> {
+    console.log(`ClusterSlots::rediscover()`, this.#runningRediscoverPromise ? 'already started rediscovering': 'start rediscovering');
     this.#runningRediscoverPromise ??= this.#rediscover(startWith)
-      .finally(() => this.#runningRediscoverPromise = undefined);
+      .finally(() => {
+        console.log(`clean rediscovery promise`);
+        this.#runningRediscoverPromise = undefined
+      });
     return this.#runningRediscoverPromise;
   }
 
@@ -550,6 +599,7 @@ export default class RedisClusterSlots<
   }
 
   async #initiatePubSubClient(toResubscribe?: PubSubToResubscribe) {
+    console.log("CS::#initiatePubSubClient", toResubscribe);
     const index = Math.floor(Math.random() * (this.masters.length + this.replicas.length)),
       node = index < this.masters.length ?
         this.masters[index] :
@@ -599,8 +649,10 @@ export default class RedisClusterSlots<
   }
 
   async #initiateShardedPubSubClient(master: MasterNode<M, F, S, RESP, TYPE_MAPPING>) {
+    console.log("CS::#initiateShardedPubSubClient()", master.host);
     const client = this.#createClient(master, false)
       .on('server-sunsubscribe', async (channel, listeners) => {
+        console.log('??????????????????????????????????', channel, listeners);
         try {
           await this.rediscover(client);
           const redirectTo = await this.getShardedPubSubClient(channel);