add slotnumber to commands

Author: nkaradzhov

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

@@ -19,6 +19,11 @@ export interface CommandOptions<T = TypeMapping> {
    * Timeout for the command in milliseconds
    */
   timeout?: number;
+  /**
+   * @internal
+   * The slot the command is targeted to (if any)
+   */
+  slotNumber?: number;
 }
 
 export interface CommandToWrite extends CommandWaitingForReply {
@@ -33,6 +38,7 @@ export interface CommandToWrite extends CommandWaitingForReply {
     listener: () => unknown;
     originalTimeout: number | undefined;
   } | undefined;
+  slotNumber?: number
 }
 
 interface CommandWaitingForReply {
@@ -219,6 +225,7 @@ export default class RedisCommandsQueue {
         channelsCounter: undefined,
         typeMapping: options?.typeMapping
       };
+      value.slotNumber = options?.slotNumber
 
       // If #maintenanceCommandTimeout was explicitly set, we should
       // use it instead of the timeout provided by the command

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

@@ -453,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
     );
   }
@@ -547,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() {

packages/client/lib/cluster/index.ts

@@ -416,13 +416,16 @@ export default class RedisCluster<
     fn: (client: RedisClientType<M, F, S, RESP, TYPE_MAPPING>, opts?: ClusterCommandOptions) => Promise<T>
   ): Promise<T> {
     const maxCommandRedirections = this._options.maxCommandRedirections ?? 16;
-    let client = await this._slots.getClient(firstKey, isReadonly);
+    let { client, slotNumber } = await this._slots.getClientAndSlotNumber(firstKey, isReadonly);
     let i = 0;
 
     let myFn = fn;
 
     while (true) {
       try {
+        if(options !== undefined) {
+          options.slotNumber = slotNumber;
+        }
         return await myFn(client, options);
       } catch (err) {
         myFn = fn;
@@ -451,7 +454,9 @@ export default class RedisCluster<
 
         if (err.message.startsWith('MOVED')) {
           await this._slots.rediscover(client);
-          client = await this._slots.getClient(firstKey, isReadonly);
+          const clientAndSlot = await this._slots.getClientAndSlotNumber(firstKey, isReadonly);
+          client = clientAndSlot.client;
+          slotNumber = clientAndSlot.slotNumber;
           continue;
         }
 
@@ -485,11 +490,11 @@ export default class RedisCluster<
     type Multi = new (...args: ConstructorParameters<typeof RedisClusterMultiCommand>) => RedisClusterMultiCommandType<[], M, F, S, RESP, TYPE_MAPPING>;
     return new ((this as any).Multi as Multi)(
       async (firstKey, isReadonly, commands) => {
-        const client = await this._self._slots.getClient(firstKey, isReadonly);
+        const { client } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly);
         return client._executeMulti(commands);
       },
       async (firstKey, isReadonly, commands) => {
-        const client = await this._self._slots.getClient(firstKey, isReadonly);
+        const { client } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly);
         return client._executePipeline(commands);
       },
       routing,