feat(client): wire multi_shard splitter into cluster dispatch

Make the multi_shard request policy actually split commands per hash
slot instead of sending the full command to one client per key.

- Routers now return a plan (RoutedCommand[]) rather than bare clients:
  pass-through policies set `client`; routeMultiShard calls the splitter
  and builds a sub-parser per slot (buildSubParser marks keys at the
  splitter's new keyPositions so `firstKey` routes each sub-command).
- The engine runs each plan entry through core `_execute` with its own
  sub-parser, so every shard gets its own sub-args; MOVED/ASK use the
  sub-command's key.
- Response reducers gain optional positionHints; reduceDefaultKeyed
  scatters split replies by groupIndices so MGET preserves caller key
  order across slots. Aggregating reducers (agg_sum, all_succeeded)
  ignore the hint.
- Splitter SubCommand now carries keyPositions for sub-parser key
  marking.

Also merge _executePolicyPlan into _executeWithPolicies — the old name
implied a plan object that never existed; it is now the single engine
(resolve policy -> plan -> execute -> reduce).

Router/reducer types are intentionally non-generic (routing sits below
the typed command surface; clients are opaque pass-through), erased to
the base cluster types instead of `any`; the engine re-narrows the
client at the `_execute` boundary.

Working end-to-end: DEL/UNLINK/EXISTS/TOUCH, MSET/MSETEX, MGET. Raw
sendCommand multi_shard and cluster-docker integration tests remain.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: nkaradzhov

packages/client/lib/cluster/index.ts

@@ -1,6 +1,6 @@
 import { RedisClientOptions, RedisClientType, WithFunctions, WithModules, WithScripts } from '../client';
 import { CommandOptions } from '../client/commands-queue';
-import { Command, CommandArguments, CommanderConfig, CommandSignature, TypeMapping, RedisArgument, RedisFunction, RedisFunctions, RedisModules, RedisScript, RedisScripts, ReplyUnion, RespVersions, TransformReply } from '../RESP/types';
+import { Command, CommandArguments, CommanderConfig, CommandSignature, TypeMapping, RedisArgument, RedisFunction, RedisFunctions, RedisModules, RedisScript, RedisScripts, ReplyUnion, RespVersions } from '../RESP/types';
 import { NON_STICKY_COMMANDS } from '../commands';
 import { EventEmitter } from 'node:events';
 import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumentsPrefix } from '../commander';
@@ -166,14 +166,14 @@ export default class RedisCluster<
     const transformReply = getTransformReply(command, resp);
 
     return async function (this: ProxyCluster, ...args: Array<unknown>) {
-      const parser = new BasicCommandParser();
+     const parser = new BasicCommandParser();
       command.parseCommand(parser, ...args);
 
       return this._self._executeWithPolicies(
         parser,
+        command.IS_READ_ONLY,
         this._commandOptions,
-        command,
-        transformReply
+        p => (client, opts) => client._executeCommand(command, p, opts, transformReply)
       );
     };
   }
@@ -187,9 +187,9 @@ export default class RedisCluster<
 
       return this._self._executeWithPolicies(
         parser,
+        command.IS_READ_ONLY,
         this._self._commandOptions,
-        command,
-        transformReply
+        p => (client, opts) => client._executeCommand(command, p, opts, transformReply)
       );
     };
   }
@@ -205,9 +205,9 @@ export default class RedisCluster<
 
       return this._self._executeWithPolicies(
         parser,
+        fn.IS_READ_ONLY,
         this._self._commandOptions,
-        fn,
-        transformReply
+        p => (client, opts) => client._executeCommand(fn, p, opts, transformReply)
       );
     };
   }
@@ -221,11 +221,11 @@ export default class RedisCluster<
       parser.push(...prefix);
       script.parseCommand(parser, ...args);
 
-      return this._self._executeScriptWithPolicies(
+      return this._self._executeWithPolicies(
         parser,
+        script.IS_READ_ONLY,
         this._commandOptions,
-        script,
-        transformReply
+        p => (client, opts) => client._executeScript(script, p, opts, transformReply)
       );
     };
   }
@@ -451,37 +451,10 @@ export default class RedisCluster<
    * Resolves the command's policies and executes it accordingly: the request
    * policy picks the target clients, each target runs through the core
    * `_execute` transport primitive, and the response policy aggregates the
-   * replies. Call sites pass data — fn closures are created here.
+   * replies. Call sites pass a `makeFn` factory that builds the per-client
+   * execution closure (command, script, or raw `sendCommand`).
    */
   async _executeWithPolicies<T>(
-    parser: CommandParser,
-    options: ClusterCommandOptions | undefined,
-    command: Command | RedisFunction,
-    transformReply: TransformReply | undefined
-  ): Promise<T> {
-    return this._executePolicyPlan(
-      parser,
-      command.IS_READ_ONLY,
-      options,
-      p => (client, opts) => client._executeCommand(command, p, opts as CommandOptions<TYPE_MAPPING> | undefined, transformReply)
-    );
-  }
-
-  async _executeScriptWithPolicies<T>(
-    parser: CommandParser,
-    options: ClusterCommandOptions | undefined,
-    script: RedisScript,
-    transformReply: TransformReply | undefined
-  ): Promise<T> {
-    return this._executePolicyPlan(
-      parser,
-      script.IS_READ_ONLY,
-      options,
-      p => (client, opts) => client._executeScript(script, p, opts as CommandOptions<TYPE_MAPPING> | undefined, transformReply)
-    );
-  }
-
-  async _executePolicyPlan<T>(
     parser: CommandParser,
     isReadonly: boolean | undefined,
     options: ClusterCommandOptions | undefined,
@@ -503,29 +476,39 @@ export default class RedisCluster<
     if (!router) {
       throw new Error(`Unknown request policy ${requestPolicy}`);
     }
-    const clients: Array<RedisClientType<M, F, S, RESP, TYPE_MAPPING>> =
-      await router(this._slots, parser, isReadonly);
+    // Routers are typed against the erased base cluster types (routing is
+    // below the typed command surface); bridge this instantiation's slots in.
+    const plan = await router(
+      this._slots as unknown as Parameters<typeof router>[0],
+      parser,
+      isReadonly,
+      policyResult.value.keySpecs
+    );
 
-    if (clients.length === 0) {
+    if (plan.length === 0) {
       throw new Error(`Request policy ${requestPolicy} produced no target nodes`);
     }
 
-    const responsePromises = clients.map(
-      client => this._execute(parser, isReadonly, options, makeFn(parser), client)
-    );
+    const responsePromises = plan.map(entry => {
+      const entryParser = entry.parser ?? parser;
+      // Re-narrow the opaque routed client to this cluster's instantiation.
+      const client = entry.client as RedisClientType<M, F, S, RESP, TYPE_MAPPING> | undefined;
+      return this._execute(entryParser, isReadonly, options, makeFn(entryParser), client);
+    });
 
     const reducer = RESPONSE_REDUCERS[responsePolicy];
     if (!reducer) {
       throw new Error(`Unknown response policy ${responsePolicy}`);
     }
-    return reducer(responsePromises, parser) as Promise<T>;
+    const positionHints = plan.map(entry => entry.groupIndices);
+    return reducer(responsePromises, parser, positionHints) as Promise<T>;
   }
 
   /**
    * Core transport primitive: sends one command to one client — resolved by
    * the parser's first key unless `pinnedClient` is given — with MOVED/ASK
    * redirect handling. Policy-free; fan-out and aggregation live in
-   * `_executePolicyPlan`.
+   * `_executeWithPolicies`.
    */
   async _execute<T>(
     parser: CommandParser,
@@ -621,12 +604,12 @@ export default class RedisCluster<
     }
 
     const parser = new BasicCommandParser();
-    firstKey && parser.push(firstKey)
+    if (firstKey) parser.push(firstKey);
     args.forEach(arg => parser.push(arg));
 
     // Raw path: no command object, so readonly-ness stays an explicit caller
     // argument and the reply is returned untransformed.
-    return this._self._executePolicyPlan(
+    return this._self._executeWithPolicies(
       parser,
       isReadonly,
       opts,

packages/client/lib/cluster/request-response-policies/dispatch.spec.ts

@@ -0,0 +1,76 @@
+import { strict as assert } from 'node:assert';
+import type { CommandParser } from '../../client/parser';
+import { reduceDefaultKeyed } from './dispatch';
+
+// The reducer ignores the parser; a stub keeps the calls readable.
+const PARSER = {} as CommandParser;
+
+describe('reduceDefaultKeyed', () => {
+  it('passes the sole reply through when not split (no hints)', async () => {
+    const reply = await reduceDefaultKeyed([Promise.resolve(['v1', 'v2'])], PARSER);
+    assert.deepEqual(reply, ['v1', 'v2']);
+  });
+
+  it('passes through when hints are all undefined (single-key command)', async () => {
+    const reply = await reduceDefaultKeyed([Promise.resolve('v1')], PARSER, [undefined]);
+    assert.equal(reply, 'v1');
+  });
+
+  it('passes through a single-slot multi_shard reply unchanged', async () => {
+    const reply = await reduceDefaultKeyed(
+      [Promise.resolve(['v0', 'v1', 'v2'])],
+      PARSER,
+      [[0, 1, 2]]
+    );
+    assert.deepEqual(reply, ['v0', 'v1', 'v2']);
+  });
+
+  it('scatters interleaved sub-replies back into original key order (MGET A,B,A,B)', async () => {
+    // keys hash to A,B,A,B -> slot A holds groups [0,2], slot B holds [1,3].
+    const reply = await reduceDefaultKeyed(
+      [
+        Promise.resolve(['a0', 'a2']),
+        Promise.resolve(['b1', 'b3'])
+      ],
+      PARSER,
+      [[0, 2], [1, 3]]
+    );
+    assert.deepEqual(reply, ['a0', 'b1', 'a2', 'b3']);
+  });
+
+  it('places each sub-reply by its hint regardless of plan order', async () => {
+    // slot B (groups [1]) listed before slot A (groups [0,2]).
+    const reply = await reduceDefaultKeyed(
+      [
+        Promise.resolve(['b1']),
+        Promise.resolve(['a0', 'a2'])
+      ],
+      PARSER,
+      [[1], [0, 2]]
+    );
+    assert.deepEqual(reply, ['a0', 'b1', 'a2']);
+  });
+
+  it('preserves null replies (missing keys) at their positions', async () => {
+    const reply = await reduceDefaultKeyed(
+      [
+        Promise.resolve(['a0', null]),
+        Promise.resolve([null])
+      ],
+      PARSER,
+      [[0, 2], [1]]
+    );
+    assert.deepEqual(reply, ['a0', null, null]);
+  });
+
+  it('throws when a split reply is missing its position hint', async () => {
+    await assert.rejects(
+      reduceDefaultKeyed(
+        [Promise.resolve(['a0']), Promise.resolve(['b1'])],
+        PARSER,
+        [[0], undefined]
+      ),
+      /missing position hints/
+    );
+  });
+});