From 42e36dfbb1b342bb46a2d72f7cfaae47866aaa9c Mon Sep 17 00:00:00 2001
From: leibale <leibale1998@gmail.com>
Date: Mon, 22 Nov 2021 17:42:41 -0500
Subject: [PATCH] enhance cluster reshard handling

---
 packages/client/lib/cluster/cluster-slots.ts  | 62 ++++++++++++-------
 packages/client/lib/cluster/index.ts          |  4 +-
 .../client/lib/commands/CLUSTER_NODES.spec.ts | 25 ++++++++
 packages/client/lib/commands/CLUSTER_NODES.ts | 13 +++-
 4 files changed, 76 insertions(+), 28 deletions(-)

diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts
index ff4c79b4d3..f69449efa1 100644
--- a/packages/client/lib/cluster/cluster-slots.ts
+++ b/packages/client/lib/cluster/cluster-slots.ts
@@ -42,20 +42,8 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
         throw new Error('None of the root nodes is available');
     }
 
-    async discover(startWith: RedisClientType<M, S>): Promise<void> {
-        if (await this.#discoverNodes(startWith.options)) return;
-
-        for (const { client } of this.#nodeByUrl.values()) {
-            if (client === startWith) continue;
-
-            if (await this.#discoverNodes(client.options)) return;
-        }
-
-        throw new Error('None of the cluster nodes is available');
-    }
-
     async #discoverNodes(clientOptions?: RedisClusterClientOptions): Promise<boolean> {
-        const client = new this.#Client(clientOptions);
+        const client = this.#initiateClient(clientOptions);
 
         await client.connect();
 
@@ -72,6 +60,29 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
         }
     }
 
+    #runningRediscoverPromise?: Promise<void>;
+
+    async rediscover(startWith: RedisClientType<M, S>): Promise<void> {
+        if (!this.#runningRediscoverPromise) {
+            this.#runningRediscoverPromise = this.#rediscover(startWith)
+                .finally(() => this.#runningRediscoverPromise = undefined);
+        }
+
+        return this.#runningRediscoverPromise;
+    }
+
+    async #rediscover(startWith: RedisClientType<M, S>): Promise<void> {
+        if (await this.#discoverNodes(startWith.options)) return;
+
+        for (const { client } of this.#nodeByUrl.values()) {
+            if (client === startWith) continue;
+
+            if (await this.#discoverNodes(client.options)) return;
+        }
+
+        throw new Error('None of the cluster nodes is available');
+    }
+
     async #reset(masters: Array<RedisClusterMasterNode>): Promise<void> {
         // Override this.#slots and add not existing clients to this.#nodeByUrl
         const promises: Array<Promise<void>> = [],
@@ -103,18 +114,23 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
         await Promise.all(promises);
     }
 
-    #clientOptionsDefaults(options: RedisClusterClientOptions): RedisClusterClientOptions {
+    #clientOptionsDefaults(options?: RedisClusterClientOptions): RedisClusterClientOptions | undefined {
         if (!this.#options.defaults) return options;
 
         const merged = Object.assign({}, this.#options.defaults, options);
 
-        if (options.socket && this.#options.defaults.socket) {
+        if (options?.socket && this.#options.defaults.socket) {
             Object.assign({}, this.#options.defaults.socket, options.socket);
         }
 
         return merged;
     }
 
+    #initiateClient(options?: RedisClusterClientOptions): RedisClientType<M, S> {
+        return new this.#Client(this.#clientOptionsDefaults(options))
+            .on('error', this.#onError);
+    }
+
     #initiateClientForNode(nodeData: RedisClusterMasterNode | RedisClusterReplicaNode, readonly: boolean, clientsInUse: Set<string>, promises: Array<Promise<void>>): ClusterNode<M, S> {
         const url = `${nodeData.host}:${nodeData.port}`;
         clientsInUse.add(url);
@@ -123,15 +139,13 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
         if (!node) {
             node = {
                 id: nodeData.id,
-                client: new this.#Client(
-                    this.#clientOptionsDefaults({
-                        socket: {
-                            host: nodeData.host,
-                            port: nodeData.port
-                        },
-                        readonly
-                    })
-                )
+                client: this.#initiateClient({
+                    socket: {
+                        host: nodeData.host,
+                        port: nodeData.port
+                    },
+                    readonly
+                })
             };
             promises.push(node.client.connect());
             this.#nodeByUrl.set(url, node);
diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts
index fcf6d4754b..11b9c937cf 100644
--- a/packages/client/lib/cluster/index.ts
+++ b/packages/client/lib/cluster/index.ts
@@ -157,7 +157,7 @@ export default class RedisCluster<M extends RedisModules = Record<string, never>
             const url = err.message.substring(err.message.lastIndexOf(' ') + 1);
             let node = this.#slots.getNodeByUrl(url);
             if (!node) {
-                await this.#slots.discover(client);
+                await this.#slots.rediscover(client);
                 node = this.#slots.getNodeByUrl(url);
 
                 if (!node) {
@@ -168,7 +168,7 @@ export default class RedisCluster<M extends RedisModules = Record<string, never>
             await node.client.asking();
             return node.client;
         } else if (err.message.startsWith('MOVED')) {
-            await this.#slots.discover(client);
+            await this.#slots.rediscover(client);
             return true;
         }
 
diff --git a/packages/client/lib/commands/CLUSTER_NODES.spec.ts b/packages/client/lib/commands/CLUSTER_NODES.spec.ts
index 2b3881d8cd..d061c59e8e 100644
--- a/packages/client/lib/commands/CLUSTER_NODES.spec.ts
+++ b/packages/client/lib/commands/CLUSTER_NODES.spec.ts
@@ -48,6 +48,31 @@ describe('CLUSTER NODES', () => {
             );
         });
 
+        it('should support urls without cport', () => {
+            assert.deepEqual(
+                transformReply(
+                    'id 127.0.0.1:30001 master - 0 0 0 connected 0-16384\n'
+                ),
+                [{
+                    id: 'id',
+                    url: '127.0.0.1:30001',
+                    host: '127.0.0.1',
+                    port: 30001,
+                    cport: null,
+                    flags: ['master'],
+                    pingSent: 0,
+                    pongRecv: 0,
+                    configEpoch: 0,
+                    linkState: RedisClusterNodeLinkStates.CONNECTED,
+                    slots: [{
+                        from: 0,
+                        to: 16384
+                    }],
+                    replicas: []
+                }]
+            );
+        });
+
         it.skip('with importing slots', () => {
             assert.deepEqual(
                 transformReply(
diff --git a/packages/client/lib/commands/CLUSTER_NODES.ts b/packages/client/lib/commands/CLUSTER_NODES.ts
index d04ffc10a1..ba4477cdd2 100644
--- a/packages/client/lib/commands/CLUSTER_NODES.ts
+++ b/packages/client/lib/commands/CLUSTER_NODES.ts
@@ -10,7 +10,7 @@ export enum RedisClusterNodeLinkStates {
 interface RedisClusterNodeTransformedUrl {
     host: string;
     port: number;
-    cport: number;
+    cport: number | null;
 }
 
 export interface RedisClusterReplicaNode extends RedisClusterNodeTransformedUrl {
@@ -86,7 +86,16 @@ export function transformReply(reply: string): Array<RedisClusterMasterNode> {
 
 function transformNodeUrl(url: string): RedisClusterNodeTransformedUrl {
     const indexOfColon = url.indexOf(':'),
-        indexOfAt = url.indexOf('@', indexOfColon);
+        indexOfAt = url.indexOf('@', indexOfColon),
+        host = url.substring(0, indexOfColon);
+
+    if (indexOfAt === -1) {
+        return {
+            host,
+            port: Number(url.substring(indexOfColon + 1)),
+            cport: null
+        };
+    }
 
     return {
         host: url.substring(0, indexOfColon),