feat(cluster): add getNodeClientForKey for WATCH/MULTI/EXEC

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.
+

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

@@ -819,6 +819,18 @@ export default class RedisClusterSlots<
     };
   }
 
+  getClientForKey(
+    key: RedisArgument,
+    isReadonly: boolean | undefined
+  ): Promise<RedisClientType<M, F, S, RESP, TYPE_MAPPING>> {
+    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));

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);

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