deduplicate smigrated notifications

Author: nkaradzhov

packages/client/lib/client/enterprise-maintenance-manager.ts

@@ -12,6 +12,7 @@ type RedisType = RedisClient<any, any, any, any, any>;
 
 export const SMIGRATED_EVENT = "__SMIGRATED";
 export interface SMigratedEvent {
+  seqId: number,
   source: { host: string, port: number };
   destination: { host: string, port: number };
   ranges: (number | [number, number])[]

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

@@ -114,6 +114,7 @@ export default class RedisClusterSlots<
   readonly nodeByAddress = new Map<string, MasterNode<M, F, S, RESP, TYPE_MAPPING> | ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
   pubSubNode?: PubSubNode<M, F, S, RESP, TYPE_MAPPING>;
   clientSideCache?: PooledClientSideCacheProvider;
+  smigratedSeqIdsSeen = new Set<number>;
 
   #isOpen = false;
 
@@ -256,6 +257,12 @@ export default class RedisClusterSlots<
   #handleSmigrated = async (event: SMigratedEvent) => {
     dbgMaintenance(`[CSlots]: handle smigrated`, event);
 
+    if(this.smigratedSeqIdsSeen.has(event.seqId)) {
+      dbgMaintenance(`[CSlots]: sequence id ${event.seqId} already seen, abort`)
+      return
+    }
+    this.smigratedSeqIdsSeen.add(event.seqId);
+
     // slots = new Array<Shard<M, F, S, RESP, TYPE_MAPPING>>(RedisClusterSlots.#SLOTS);
     // masters = new Array<MasterNode<M, F, S, RESP, TYPE_MAPPING>>();
     // replicas = new Array<ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
@@ -446,7 +453,7 @@ export default class RedisClusterSlots<
   nodeClient(node: ShardNode<M, F, S, RESP, TYPE_MAPPING>) {
     return (
       node.connectPromise ?? // if the node is connecting
-      node.client ?? // if the node is connected
+      (node.client ? Promise.resolve(node.client) : undefined) ?? // if the node is connected
       this.#createNodeClient(node) // if the not is disconnected
     );
   }
@@ -540,20 +547,30 @@ export default class RedisClusterSlots<
     this.#emit('disconnect');
   }
 
-  getClient(
+  async getClientAndSlotNumber(
     firstKey: RedisArgument | undefined,
     isReadonly: boolean | undefined
-  ) {
+  ): Promise<{
+    client: RedisClientType<M, F, S, RESP, TYPE_MAPPING>,
+    slotNumber?: number
+  }> {
     if (!firstKey) {
-      return this.nodeClient(this.getRandomNode());
+      return {
+        client: await this.nodeClient(this.getRandomNode())
+      };
     }
 
     const slotNumber = calculateSlot(firstKey);
     if (!isReadonly) {
-      return this.nodeClient(this.slots[slotNumber].master);
+      return {
+        client: await this.nodeClient(this.slots[slotNumber].master),
+        slotNumber
+      };
     }
 
-    return this.nodeClient(this.getSlotRandomNode(slotNumber));
+    return {
+      client: await this.nodeClient(this.getSlotRandomNode(slotNumber))
+    };
   }
 
   *#iterateAllNodes() {