feat(client): add MaxCommandRedirectionsError for cluster redirections

packages/client/lib/cluster/index.spec.ts

@@ -3,7 +3,7 @@ import { setTimeout } from 'node:timers/promises';
 import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
 import RedisCluster from '.';
 import { SQUARE_SCRIPT } from '../client/index.spec';
-import { ClientClosedError, ClientOfflineError, RootNodesUnavailableError } from '../errors';
+import { ClientClosedError, ClientOfflineError, RootNodesUnavailableError, MaxCommandRedirectionsError } from '../errors';
 import { spy } from 'sinon';
 import RedisClient from '../client';
 import { RESP_TYPES } from '../RESP/decoder';
@@ -243,6 +243,79 @@ describe('Cluster', () => {
     numberOfMasters: 2
   });
 
+  testUtils.testWithCluster('should throw MaxCommandRedirectionsError when exceeding maxCommandRedirections', async cluster => {
+    const key = 'maxCommandRedirectionsErrorKey',
+      slot = calculateSlot(key),
+      node1 = cluster.masters[0],
+      node2 = cluster.masters[1],
+      [client1, client2] = await Promise.all([
+        cluster.nodeClient(node1),
+        cluster.nodeClient(node2)
+      ]);
+
+    // Create a MOVED loop
+    await Promise.all([
+      client1.clusterSetSlot(slot, 'NODE', node2.id),
+      client2.clusterSetSlot(slot, 'NODE', node1.id)
+    ]);
+
+    try {
+      await assert.rejects(
+        cluster.get(key),
+        MaxCommandRedirectionsError
+      );
+    } finally {
+      // Revert the slot back to its original owner (node1)
+      await Promise.all([
+        client1.clusterSetSlot(slot, 'NODE', node1.id),
+        client2.clusterSetSlot(slot, 'NODE', node1.id)
+      ]);
+    }
+  }, {
+    serverArguments: [],
+    numberOfMasters: 2,
+    clusterConfiguration: {
+      maxCommandRedirections: 2
+    }
+  });
+
+  testUtils.testWithCluster('sSubscribe should throw MaxCommandRedirectionsError when exceeding maxCommandRedirections', async cluster => {
+    const channel = 'maxCommandRedirectionsErrorChannel',
+      slot = calculateSlot(channel),
+      node1 = cluster.masters[0],
+      node2 = cluster.masters[1],
+      [client1, client2] = await Promise.all([
+        cluster.nodeClient(node1),
+        cluster.nodeClient(node2)
+      ]);
+
+    // Create a MOVED loop
+    await Promise.all([
+      client1.clusterSetSlot(slot, 'NODE', node2.id),
+      client2.clusterSetSlot(slot, 'NODE', node1.id)
+    ]);
+
+    try {
+      await assert.rejects(
+        cluster.sSubscribe(channel, () => {}),
+        MaxCommandRedirectionsError
+      );
+    } finally {
+      // Revert the slot back to its original owner (node1)
+      await Promise.all([
+        client1.clusterSetSlot(slot, 'NODE', node1.id),
+        client2.clusterSetSlot(slot, 'NODE', node1.id)
+      ]);
+    }
+  }, {
+    serverArguments: [],
+    numberOfMasters: 2,
+    clusterConfiguration: {
+      maxCommandRedirections: 2
+    },
+    minimumDockerVersion: [7]
+  });
+
   testUtils.testWithCluster('getRandomNode should spread the the load evenly', async cluster => {
     const totalNodes = cluster.masters.length + cluster.replicas.length,
       ids = new Set<string>();

packages/client/lib/cluster/index.ts

@@ -7,7 +7,7 @@ import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumen
 import RedisClusterSlots, { NodeAddressMap, RESUBSCRIBE_LISTENERS_EVENT, ShardNode } from './cluster-slots';
 import RedisClusterMultiCommand, { RedisClusterMultiCommandType } from './multi-command';
 import { PubSubListener, PubSubListeners } from '../client/pub-sub';
-import { ErrorReply } from '../errors';
+import { ErrorReply, MaxCommandRedirectionsError } from '../errors';
 import { RedisTcpSocketOptions } from '../client/socket';
 import { ClientSideCacheConfig, PooledClientSideCacheProvider } from '../client/cache';
 import { BasicCommandParser, CommandParser } from '../client/parser';
@@ -714,16 +714,23 @@ export default class RedisCluster<
           const err = _err as Error;
           myFn = fn;
 
-          // TODO: error class
-          if (++i > maxCommandRedirections || !(err instanceof Error)) {
-            if (err instanceof Error) {
-              publish(CHANNELS.ERROR, () => ({
-                error: err,
-                origin: 'cluster',
-                internal: false,
-                clientId: client._clientId,
-                retryCount: i,
-              }));
+          if (!(err instanceof Error)) {
+            throw err;
+          }
+
+          const isRedirect = err.message.startsWith('ASK') || err.message.startsWith('MOVED');
+
+          if (++i > maxCommandRedirections) {
+            publish(CHANNELS.ERROR, () => ({
+              error: err,
+              origin: 'cluster',
+              internal: false,
+              clientId: client._clientId,
+              retryCount: i,
+            }));
+
+            if (isRedirect) {
+              throw new MaxCommandRedirectionsError(err);
             }
             throw err;
           }
@@ -881,11 +888,20 @@ export default class RedisCluster<
       try {
         return await client.SSUBSCRIBE(channels, listener, bufferMode);
       } catch (err) {
-        if (++i > maxCommandRedirections || !(err instanceof ErrorReply)) {
+        if (!(err instanceof ErrorReply)) {
+          throw err;
+        }
+
+        const isRedirect = err.message.startsWith('MOVED');
+
+        if (++i > maxCommandRedirections) {
+          if (isRedirect) {
+            throw new MaxCommandRedirectionsError(err);
+          }
           throw err;
         }
 
-        if (err.message.startsWith('MOVED')) {
+        if (isRedirect) {
           await this._self._slots.rediscover(client);
           client = await this._self._slots.getShardedPubSubClient(firstChannel);
           continue;

packages/client/lib/errors.ts

@@ -107,3 +107,9 @@ export class MultiErrorReply extends ErrorReply {
 }
 
 export class OpenTelemetryError extends Error { }
+
+export class MaxCommandRedirectionsError extends Error {
+  constructor(cause?: unknown) {
+    super('Too many Cluster redirections', cause === undefined ? undefined : { cause });
+  }
+}