From f3b731f86753581c84e2d314afdeb74b137a10e1 Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Thu, 25 Jun 2026 14:14:29 +0300 Subject: [PATCH] feat(cluster): add getNodeClientForKey for WATCH/MULTI/EXEC Cluster client cannot expose WATCH because it needs connection-level state on a specific node. Previously users had to import cluster-key-slot, compute the slot, read cluster.slots, and call nodeClient manually. Add getNodeClientForKey(key, isReadonly?) which resolves the key's slot and returns the connected node client (master, or a slot node when readonly), enabling optimistic-locking transactions in cluster mode. Closes #3194 Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/clustering.md | 18 +++++++++++++++++ packages/client/lib/cluster/cluster-slots.ts | 12 +++++++++++ packages/client/lib/cluster/index.spec.ts | 20 +++++++++++++++++++ packages/client/lib/cluster/index.ts | 21 ++++++++++++++++++++ 4 files changed, 71 insertions(+) diff --git a/docs/clustering.md b/docs/clustering.md index 4afd95afd23..c43d63407d0 100644 --- a/docs/clustering.md +++ b/docs/clustering.md @@ -152,3 +152,21 @@ Admin commands such as `MEMORY STATS`, `FLUSHALL`, etc. are not attached to the Certain commands (e.g. `PUBLISH`) are forwarded to other cluster nodes by the Redis server. The client sends these commands to a random node in order to spread the load across the cluster. +### Transactions with `WATCH` + +`WATCH` relies on connection-level state on a specific node, so it isn't exposed directly on the cluster client. Use `.getNodeClientForKey()` to get the node client responsible for a key's slot and run the optimistic-locking transaction on it: + +```javascript +const key = 'key'; +const nodeClient = await cluster.getNodeClientForKey(key); + +await nodeClient.watch(key); +const value = await nodeClient.get(key); +const reply = await nodeClient + .multi() + .set(key, calculateNewValue(value)) // application logic + .exec(); // `null` if `key` changed since `WATCH`, retry in that case +``` + +All keys touched in the transaction must hash to the same slot. Pass `true` as the second argument (`getNodeClientForKey(key, true)`) to allow a replica for read-only use. + diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index ac581372810..e6dc0ba103a 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -819,6 +819,18 @@ export default class RedisClusterSlots< }; } + getClientForKey( + key: RedisArgument, + isReadonly: boolean | undefined + ): Promise> { + const slotNumber = calculateSlot(key); + if (isReadonly) { + return this.nodeClient(this.getSlotRandomNode(slotNumber)); + } + + return this.nodeClient(this.slots[slotNumber].master); + } + *#iterateAllNodes() { if(this.masters.length + this.replicas.length === 0) return let i = Math.floor(Math.random() * (this.masters.length + this.replicas.length)); diff --git a/packages/client/lib/cluster/index.spec.ts b/packages/client/lib/cluster/index.spec.ts index 0360d7d66cb..1266ee59a6e 100644 --- a/packages/client/lib/cluster/index.spec.ts +++ b/packages/client/lib/cluster/index.spec.ts @@ -7,6 +7,7 @@ import { RootNodesUnavailableError } from '../errors'; import { spy } from 'sinon'; import RedisClient from '../client'; import { RESP_TYPES } from '../RESP/decoder'; +import calculateSlot from 'cluster-key-slot'; describe('Cluster', () => { describe('default commandOptions', () => { @@ -258,6 +259,25 @@ describe('Cluster', () => { } }); + testUtils.testWithCluster('getNodeClientForKey returns the slot master and supports WATCH/MULTI/EXEC', async cluster => { + const key = 'key'; + const nodeClient = await cluster.getNodeClientForKey(key); + assert.ok(nodeClient instanceof RedisClient); + assert.equal(nodeClient, cluster.slots[calculateSlot(key)].master.client); + + await nodeClient.watch(key); + const reply = await nodeClient.multi() + .set(key, 'value') + .exec(); + assert.deepEqual(reply, ['OK']); + }, GLOBAL.CLUSTERS.OPEN); + + testUtils.testWithCluster('getNodeClientForKey with isReadonly returns a node from the slot', async cluster => { + const key = 'key'; + const nodeClient = await cluster.getNodeClientForKey(key, true); + assert.ok(nodeClient instanceof RedisClient); + }, GLOBAL.CLUSTERS.WITH_REPLICAS); + testUtils.testWithCluster('should throw CROSSSLOT error', async cluster => { await assert.rejects(cluster.mGet(['a', 'b'])); }, GLOBAL.CLUSTERS.OPEN); diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts index 8035b0a7f3f..f7ba6fb1757 100644 --- a/packages/client/lib/cluster/index.ts +++ b/packages/client/lib/cluster/index.ts @@ -728,6 +728,27 @@ export default class RedisCluster< return this._self._slots.getSlotRandomNode(slot); } + /** + * Returns the connected node client responsible for the given key's slot. + * Useful for connection-level operations that the cluster client does not expose + * directly, such as `WATCH` followed by `MULTI`/`EXEC`: + * + * ```javascript + * const nodeClient = await cluster.getNodeClientForKey(key); + * await nodeClient.WATCH(key); + * const value = await nodeClient.GET(key); + * const reply = await nodeClient.MULTI() + * .SET(key, calculateNewValue(value)) + * .EXEC(); // `null` if `key` changed, retry + * ``` + * + * @param key - The key whose slot determines the node. + * @param isReadonly - If `true`, may return a replica client; otherwise returns the slot master. + */ + getNodeClientForKey(key: RedisArgument, isReadonly?: boolean) { + return this._self._slots.getClientForKey(key, isReadonly); + } + /** * @deprecated use `.masters` instead * TODO