feat(client): sticky FT.CURSOR routing via special request policy

FT.CURSOR READ/DEL carry no key, so hash-slot routing can't reach the
coordinator that minted the cursor via FT.AGGREGATE ...WITHCURSOR and the
server rejects the unknown cursor. Flip ft.cursor to the HLD `special`
request policy and add client-side sticky machinery:

- cluster-slots: per-instance cursor binding registry
  ((index,cursorId) -> address) with bind/lookup/evict + opportunistic
  idle sweep, plus nodeAddressByClient reverse-lookup.
- ft-cursor: routeFtCursor (pin bound node / throw MISS before any
  network call), SPECIAL_REQUEST_ROUTERS, extractCursorId, and the
  captureCursorBinding post-reply hook (AGGREGATE bind / READ
  rebind-refresh-evict / DEL evict).
- dispatch: routeSpecial short-circuits into SPECIAL_REQUEST_ROUTERS,
  keeps the warn-and-random fallback for unregistered special commands.
- _executeWithPolicies: best-effort capture hook after the reply
  resolves; _execute untouched.
- static policies: override ft.cursor -> special (+ READ/DEL
  subcommands, response stays default-keyless), regenerated data table.

Response stays single-node pass-through (no special reducer). Bindings
are per client instance; cross-instance/cross-process cursors MISS by
design. Unit tests cover router/capture/lifecycle/collision; cluster
integration tests cover pagination-to-completion, DEL->READ MISS, and
cross-instance MISS.

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

Author: nkaradzhov

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

@@ -22,6 +22,26 @@ export type NodeAddressMap = {
 
 export const RESUBSCRIBE_LISTENERS_EVENT = '__resubscribeListeners'
 
+/**
+ * Sticky-cursor binding: which node served a RediSearch cursor. FT.CURSOR
+ * READ/DEL carry no key, so hash-slot routing can't reach the coordinator that
+ * minted the cursor — we pin by `address` ("host:port"), the durable handle
+ * (clients are recreated on reconnect/topology refresh, addresses aren't).
+ */
+export interface CursorBinding {
+  address: string;
+  createdAt: number;
+  maxIdleMs?: number;
+}
+
+/**
+ * Fallback idle TTL for the opportunistic cursor-binding sweep when the
+ * FT.AGGREGATE didn't declare MAXIDLE. Mirrors the RediSearch default (300s)
+ * so abandoned cursors don't leak the binding map (timer-free, like
+ * `smigratedSeqIdsSeen`).
+ */
+const DEFAULT_CURSOR_MAX_IDLE_MS = 300_000;
+
 export interface Node<
   M extends RedisModules,
   F extends RedisFunctions,
@@ -121,6 +141,8 @@ export default class RedisClusterSlots<
   pubSubNode?: PubSubNode<M, F, S, RESP, TYPE_MAPPING>;
   clientSideCache?: PooledClientSideCacheProvider;
   smigratedSeqIdsSeen = new Set<number>;
+  /** Per-instance sticky-cursor bindings, keyed `${index}:${cursorId}`. */
+  readonly cursorBindings = new Map<string, CursorBinding>();
   #topologyRefreshPromise?: Promise<boolean | void>;
 
   #isOpen = false;
@@ -918,6 +940,50 @@ export default class RedisClusterSlots<
     return this.nodeClient(master);
   }
 
+  /**
+   * Reverse-resolve a routed client to its node address. FT.AGGREGATE is
+   * keyless, so the plan carries only the client; we need its address to bind
+   * the cursor. Clients are few per cluster, so the linear scan is negligible.
+   */
+  nodeAddressByClient(client: RedisClientType<M, F, S, RESP, TYPE_MAPPING>): string | undefined {
+    for (const [address, node] of this.nodeByAddress) {
+      if (node.client === client) return address;
+    }
+    return undefined;
+  }
+
+  #cursorKey(index: string, cursorId: number) {
+    return `${index}:${cursorId}`;
+  }
+
+  /**
+   * Drop bindings idle past their MAXIDLE (or the default TTL). Opportunistic —
+   * runs on each `bindCursor` so there's no timer to manage (see
+   * `smigratedSeqIdsSeen`). Cheap: the map holds only live cursors.
+   */
+  #sweepStaleCursors(now: number) {
+    for (const [key, binding] of this.cursorBindings) {
+      const ttl = binding.maxIdleMs ?? DEFAULT_CURSOR_MAX_IDLE_MS;
+      if (now - binding.createdAt > ttl) {
+        this.cursorBindings.delete(key);
+      }
+    }
+  }
+
+  bindCursor(index: string, cursorId: number, address: string, maxIdleMs?: number) {
+    const now = Date.now();
+    this.#sweepStaleCursors(now);
+    this.cursorBindings.set(this.#cursorKey(index, cursorId), { address, createdAt: now, maxIdleMs });
+  }
+
+  lookupCursor(index: string, cursorId: number): CursorBinding | undefined {
+    return this.cursorBindings.get(this.#cursorKey(index, cursorId));
+  }
+
+  evictCursor(index: string, cursorId: number) {
+    this.cursorBindings.delete(this.#cursorKey(index, cursorId));
+  }
+
   getPubSubClient(): Promise<RedisClientType<M, F, S, RESP, TYPE_MAPPING>> {
     if (!this.pubSubNode) return this.#initiatePubSubClient();
 

packages/client/lib/cluster/index.ts

@@ -18,6 +18,7 @@ import { ClientIdentity, ClientRole, generateClusterClientId } from '../client/i
 import { DEFAULT_COMMAND_TIMEOUT } from '../defaults';
 import { POLICIES, PolicyResolver, StaticPolicyResolver, REQUEST_POLICIES_WITH_DEFAULTS, RESPONSE_POLICIES_WITH_DEFAULTS, type CommandPolicies } from './request-response-policies';
 import { REQUEST_ROUTERS, RESPONSE_REDUCERS } from './request-response-policies/dispatch';
+import { captureCursorBinding } from './request-response-policies/ft-cursor';
 
 export type ClusterTopologyRefreshOnReconnectionAttemptStrategy =
   false |
@@ -535,7 +536,22 @@ export default class RedisCluster<
       throw new Error(`Unknown response policy ${responsePolicy}`);
     }
     const positionHints = plan.map(entry => entry.groupIndices);
-    return reducer(responsePromises, parser, positionHints) as Promise<T>;
+    const reply = await (reducer(responsePromises, parser, positionHints) as Promise<T>);
+
+    // Sticky-cursor bookkeeping: FT.AGGREGATE/FT.CURSOR bind/rebind/evict the
+    // serving node from the resolved reply. Command-name gated and best-effort
+    // (a bad binding only downgrades to a MISS throw on the next READ/DEL), so
+    // never let it mask the caller's reply.
+    try {
+      captureCursorBinding(
+        this._slots as unknown as Parameters<typeof captureCursorBinding>[0],
+        parser,
+        plan,
+        reply
+      );
+    } catch { /* binding capture is best-effort */ }
+
+    return reply;
   }
 
   /**

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

@@ -20,6 +20,7 @@ import {
   type RequestPolicyWithDefaults,
   type ResponsePolicyWithDefaults
 } from './policies-constants';
+import { SPECIAL_REQUEST_ROUTERS } from './ft-cursor';
 
 // Routing runs *below* the typed command surface: routers never inspect the
 // command's M/F/S/RESP/TM parameters, they just shuffle opaque clients from
@@ -121,15 +122,17 @@ function specialKey(parser: CommandParser): string {
 }
 
 /**
- * Fallback router for `special` request commands without a dedicated handler.
- * A `special` request policy means non-trivial routing that no generic rule
- * captures; we don't know the correct target, so route to a single (random)
- * node like a keyless command. This keeps the command working instead of
- * throwing, but the reply reflects only that one node — warn so the gap is
- * visible. Commands with a real handler never reach this.
+ * Router for the `special` request policy. Commands with a dedicated handler
+ * (e.g. FT.CURSOR sticky routing) short-circuit into `SPECIAL_REQUEST_ROUTERS`
+ * first. Everything else has non-trivial routing no generic rule captures and
+ * no handler yet: route to a single (random) node like a keyless command so it
+ * still works, but warn — the reply reflects only that one node.
  */
 export const routeSpecial: RequestRouter =
-  async (slots, parser) => {
+  async (slots, parser, isReadonly, keySpecs) => {
+    const handler = SPECIAL_REQUEST_ROUTERS[specialKey(parser)];
+    if (handler) return handler(slots, parser, isReadonly, keySpecs);
+
     console.warn(
       `node-redis: no cluster routing implemented for the "special" request policy of ` +
       `"${specialKey(parser)}"; routing to a single node. The reply may be incomplete.`

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

@@ -0,0 +1,163 @@
+import { strict as assert } from 'node:assert';
+import type { CommandParser } from '../../client/parser';
+import { routeFtCursor, captureCursorBinding, extractCursorId } from './ft-cursor';
+
+/**
+ * Minimal stand-in for the cursor-relevant surface of `RedisClusterSlots`,
+ * mirroring the real `${index}:${cursorId}` keying and address→client map so
+ * the router/capture logic is exercised without spinning a cluster.
+ */
+class FakeSlots {
+  cursorBindings = new Map<string, { address: string; createdAt: number; maxIdleMs?: number }>();
+  clientsByAddress = new Map<string, object>();
+
+  #key(index: string, cursorId: number) { return `${index}:${cursorId}`; }
+  bindCursor(index: string, cursorId: number, address: string, maxIdleMs?: number) {
+    this.cursorBindings.set(this.#key(index, cursorId), { address, createdAt: 0, maxIdleMs });
+  }
+  lookupCursor(index: string, cursorId: number) { return this.cursorBindings.get(this.#key(index, cursorId)); }
+  evictCursor(index: string, cursorId: number) { this.cursorBindings.delete(this.#key(index, cursorId)); }
+  async getMasterByAddress(address: string) { return this.clientsByAddress.get(address); }
+  nodeAddressByClient(client: object) {
+    for (const [address, c] of this.clientsByAddress) if (c === client) return address;
+    return undefined;
+  }
+}
+
+const parserOf = (...args: Array<string>) =>
+  ({ redisArgs: args, commandIdentifier: { command: args[0], subcommand: args[1] } }) as unknown as CommandParser;
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any -- routers/capture run below the typed surface
+const asSlots = (s: FakeSlots) => s as any;
+
+describe('extractCursorId', () => {
+  it('reads the transformed-path `{ cursor }` object (RESP2 + RESP3)', () => {
+    assert.equal(extractCursorId({ total: 1, results: [], cursor: 42 }), 42);
+  });
+
+  it('reads raw RESP2 `[result, cursor]` at index 1', () => {
+    assert.equal(extractCursorId([['result'], 7]), 7);
+  });
+
+  it('reads raw RESP3 map key `cursor`', () => {
+    assert.equal(extractCursorId(new Map<string, unknown>([['results', []], ['cursor', 9]])), 9);
+  });
+
+  it('returns undefined when there is no cursor (e.g. FT.CURSOR DEL "OK")', () => {
+    assert.equal(extractCursorId('OK'), undefined);
+    assert.equal(extractCursorId(null), undefined);
+  });
+});
+
+describe('routeFtCursor', () => {
+  it('pins the bound client on HIT', async () => {
+    const slots = new FakeSlots();
+    const client = { id: 'node-a' };
+    slots.clientsByAddress.set('127.0.0.1:7000', client);
+    slots.bindCursor('idx', 123, '127.0.0.1:7000');
+
+    const plan = await routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '123'), undefined, undefined);
+    assert.deepEqual(plan, [{ client }]);
+  });
+
+  it('throws on MISS (cursor never bound)', async () => {
+    const slots = new FakeSlots();
+    await assert.rejects(
+      routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '404'), undefined, undefined),
+      /no known node for cursor 404 on index "idx"/
+    );
+  });
+
+  it('throws when the bound node is gone (getMasterByAddress → undefined)', async () => {
+    const slots = new FakeSlots();
+    slots.bindCursor('idx', 5, '127.0.0.1:9999'); // address not in clientsByAddress
+    await assert.rejects(
+      routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'DEL', 'idx', '5'), undefined, undefined),
+      /left the cluster/
+    );
+  });
+});
+
+describe('captureCursorBinding — FT.AGGREGATE', () => {
+  it('binds (index, cursor) → serving node address (RESP2 array reply)', () => {
+    const slots = new FakeSlots();
+    const client = {};
+    slots.clientsByAddress.set('10.0.0.1:6379', client);
+
+    captureCursorBinding(asSlots(slots), parserOf('FT.AGGREGATE', 'idx', '*', 'WITHCURSOR'), [{ client } as any], [[], 55]);
+    assert.deepEqual(slots.lookupCursor('idx', 55), { address: '10.0.0.1:6379', createdAt: 0, maxIdleMs: undefined });
+  });
+
+  it('binds from the transformed `{ cursor }` reply and captures MAXIDLE', () => {
+    const slots = new FakeSlots();
+    const client = {};
+    slots.clientsByAddress.set('10.0.0.1:6379', client);
+
+    captureCursorBinding(
+      asSlots(slots),
+      parserOf('FT.AGGREGATE', 'idx', '*', 'WITHCURSOR', 'MAXIDLE', '5000'),
+      [{ client } as any],
+      { total: 0, results: [], cursor: 88 }
+    );
+    assert.deepEqual(slots.lookupCursor('idx', 88), { address: '10.0.0.1:6379', createdAt: 0, maxIdleMs: 5000 });
+  });
+
+  it('does not bind when the aggregate exhausts in one batch (cursor 0)', () => {
+    const slots = new FakeSlots();
+    const client = {};
+    slots.clientsByAddress.set('10.0.0.1:6379', client);
+
+    captureCursorBinding(asSlots(slots), parserOf('FT.AGGREGATE', 'idx', '*', 'WITHCURSOR'), [{ client } as any], { cursor: 0 });
+    assert.equal(slots.cursorBindings.size, 0);
+  });
+});
+
+describe('captureCursorBinding — FT.CURSOR lifecycle', () => {
+  const seed = () => {
+    const slots = new FakeSlots();
+    const client = {};
+    slots.clientsByAddress.set('10.0.0.1:6379', client);
+    slots.bindCursor('idx', 100, '10.0.0.1:6379');
+    return { slots, client };
+  };
+
+  it('rebinds a continuation cursor (evict old, bind new, same address)', () => {
+    const { slots, client } = seed();
+    captureCursorBinding(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '100'), [{ client } as any], { cursor: 200 });
+    assert.equal(slots.lookupCursor('idx', 100), undefined);
+    assert.deepEqual(slots.lookupCursor('idx', 200), { address: '10.0.0.1:6379', createdAt: 0, maxIdleMs: undefined });
+  });
+
+  it('evicts on READ → cursor 0 (exhausted)', () => {
+    const { slots, client } = seed();
+    captureCursorBinding(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '100'), [{ client } as any], { cursor: 0 });
+    assert.equal(slots.lookupCursor('idx', 100), undefined);
+  });
+
+  it('keeps the binding when the continuation id is unchanged', () => {
+    const { slots, client } = seed();
+    captureCursorBinding(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '100'), [{ client } as any], { cursor: 100 });
+    assert.deepEqual(slots.lookupCursor('idx', 100), { address: '10.0.0.1:6379', createdAt: 0, maxIdleMs: undefined });
+  });
+
+  it('evicts on DEL regardless of reply, then a follow-up READ MISSes', async () => {
+    const { slots, client } = seed();
+    captureCursorBinding(asSlots(slots), parserOf('FT.CURSOR', 'DEL', 'idx', '100'), [{ client } as any], 'OK');
+    assert.equal(slots.lookupCursor('idx', 100), undefined);
+    await assert.rejects(routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idx', '100'), undefined, undefined));
+  });
+});
+
+describe('cursor-id collision across indexes', () => {
+  it('keys on (index, cursorId) so same id under two indexes routes independently', async () => {
+    const slots = new FakeSlots();
+    const clientA = { id: 'a' }, clientB = { id: 'b' };
+    slots.clientsByAddress.set('a:1', clientA);
+    slots.clientsByAddress.set('b:1', clientB);
+    slots.bindCursor('idxA', 1, 'a:1');
+    slots.bindCursor('idxB', 1, 'b:1');
+
+    assert.deepEqual(await routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idxA', '1'), undefined, undefined), [{ client: clientA }]);
+    assert.deepEqual(await routeFtCursor(asSlots(slots), parserOf('FT.CURSOR', 'READ', 'idxB', '1'), undefined, undefined), [{ client: clientB }]);
+  });
+});