diff --git a/packages/client/lib/cluster/index.spec.ts b/packages/client/lib/cluster/index.spec.ts index ede3ff5c53e..527913d6009 100644 --- a/packages/client/lib/cluster/index.spec.ts +++ b/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(); diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts index 5467381aad0..a10c1baa1db 100644 --- a/packages/client/lib/cluster/index.ts +++ b/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; diff --git a/packages/client/lib/errors.ts b/packages/client/lib/errors.ts index 472a8ae5cc8..952b565f83d 100644 --- a/packages/client/lib/errors.ts +++ b/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 }); + } +}