feat(client): add HIMPORT command family with managed fieldset lifecycle

packages/client/lib/client/commands-queue.ts

@@ -24,6 +24,13 @@ export interface CommandOptions<T = TypeMapping> {
    * The slot the command is targeted to (if any)
    */
   slotNumber?: number;
+  /**
+   * @internal
+   * Set by the cluster ASK-redirect handler. Signals the HIMPORT hook that this command rides
+   * an ASK chain, so any keyless command it injects ahead of the main (key-bearing) command
+   * would consume the one-shot ASKING flag — the hook re-issues ASKING right before the main.
+   */
+  askRedirect?: boolean;
 }
 
 export interface CommandToWrite extends CommandWaitingForReply {

packages/client/lib/client/pool.ts

@@ -14,6 +14,7 @@ import SingleEntryCache from '../single-entry-cache';
 import { MULTI_MODE, MultiMode } from '../multi-command';
 import { publish, CHANNELS } from './tracing';
 import { ClientIdentity, ClientRole, generateClientId } from './identity';
+import { FieldsetRegistry } from '../himport/registry';
 
 export interface RedisPoolOptions {
   /**
@@ -324,6 +325,12 @@ export class RedisClientPool<
       }
     }
 
+    // One fieldset registry for the whole pool: a fieldset registered through any borrowed
+    // connection must be transparently re-preparable on every other pooled connection.
+    // Deliberately NOT inherited when the pool is created from an existing client
+    // (createPool spreads the client's options, which never carry an instance).
+    clientOptions = { ...clientOptions, himportRegistry: new FieldsetRegistry() };
+
     // Capture the key prefix for the pool's own parser construction, then strip it from
     // the pooled clients' options: the pool builds the (already prefixed) parser and hands
     // it to a pooled client, which never re-parses — so inner clients must not re-prefix.

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

@@ -10,6 +10,7 @@ import { BasicPooledClientSideCache, PooledClientSideCacheProvider } from '../cl
 import { SMIGRATED_EVENT, SMigratedEvent, dbgMaintenance } from '../client/enterprise-maintenance-manager';
 import { ClientRole } from '../client/identity';
 import ClusterReconnectionTracker from './cluster-reconnection-tracker';
+import { FieldsetRegistry } from '../himport/registry';
 
 interface NodeAddress {
   host: string;
@@ -120,6 +121,17 @@ export default class RedisClusterSlots<
   readonly nodeByAddress = new Map<string, MasterNode<M, F, S, RESP, TYPE_MAPPING> | ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
   pubSubNode?: PubSubNode<M, F, S, RESP, TYPE_MAPPING>;
   clientSideCache?: PooledClientSideCacheProvider;
+  /**
+   * One fieldset registry for the whole cluster: HIMPORT PREPARE fans out to all masters
+   * via the policy layer, and every node client (including MOVED/rediscovered nodes and
+   * SMIGRATED destinations) must lazily re-prepare from the same registrations.
+   */
+  readonly #himportRegistry: FieldsetRegistry;
+
+  /** The cluster-wide registry, exposed so `RedisCluster.duplicate()` can share it. */
+  get himportRegistry() {
+    return this.#himportRegistry;
+  }
   smigratedSeqIdsSeen = new Set<number>;
   #topologyRefreshPromise?: Promise<boolean | void>;
 
@@ -148,6 +160,7 @@ export default class RedisClusterSlots<
   ) {
     this.#validateOptions(options);
     this.#options = options;
+    this.#himportRegistry = options.himportRegistry ?? new FieldsetRegistry();
     this.#clusterClientId = clusterClientId;
     this.#reconnectionTracker = new ClusterReconnectionTracker(options.topologyRefreshOnReconnectionAttemptStrategy);
 
@@ -603,6 +616,7 @@ export default class RedisClusterSlots<
     let wasReady = false;
     const client = this.#clientFactory( this.#clientOptionsDefaults({
         clientSideCache: this.clientSideCache,
+        himportRegistry: this.#himportRegistry,
         RESP: this.#options.RESP,
         socket,
         readonly,

packages/client/lib/cluster/index.ts

@@ -16,6 +16,7 @@ import SingleEntryCache from '../single-entry-cache'
 import { publish, CHANNELS } from '../client/tracing';
 import { ClientIdentity, ClientRole, generateClusterClientId } from '../client/identity';
 import { DEFAULT_COMMAND_TIMEOUT } from '../defaults';
+import { FieldsetRegistry } from '../himport/registry';
 
 export type ClusterTopologyRefreshOnReconnectionAttemptStrategy =
   false |
@@ -146,6 +147,14 @@ export interface RedisClusterOptions<
    * ```
    */
   clientSideCache?: PooledClientSideCacheProvider | ClientSideCacheConfig;
+  /**
+   * @internal
+   * Shared HIMPORT fieldset registry for the whole cluster. When omitted the cluster's
+   * `RedisClusterSlots` owns a fresh one; `duplicate()` injects the parent's so the duplicate
+   * SHARES the parent's registrations (matching the standalone `duplicate()` guarantee).
+   * Not a user-facing option.
+   */
+  himportRegistry?: FieldsetRegistry;
 }
 
 export type RedisClusterType<
@@ -410,6 +419,9 @@ export default class RedisCluster<
   >(overrides?: Partial<RedisClusterOptions<_M, _F, _S, _RESP, _TYPE_MAPPING>>) {
     return new (Object.getPrototypeOf(this).constructor)({
       ...this._self._options,
+      // Inject the live registry explicitly (the options spread only carries it if the user
+      // passed one) — a duplicate SHARES the parent's HIMPORT registrations.
+      himportRegistry: this._self._slots.himportRegistry,
       commandOptions: this._commandOptions,
       ...overrides
     }) as RedisClusterType<_M, _F, _S, _RESP, _TYPE_MAPPING>;
@@ -478,6 +490,9 @@ export default class RedisCluster<
       const chainId = Symbol("asking chain");
       const opts = options ? {...options} : {};
       opts.chainId = chainId;
+      // Tell the HIMPORT hook it is on an ASK chain: a keyless PREPARE/DISCARD it pipelines
+      // ahead of the SET would eat the one-shot ASKING flag, so the hook re-issues ASKING.
+      opts.askRedirect = true;
 
 
 

packages/client/lib/commands/HIMPORT_DISCARD.spec.ts

@@ -0,0 +1,36 @@
+import { strict as assert } from 'node:assert';
+import testUtils, { GLOBAL } from '../test-utils';
+import HIMPORT_DISCARD from './HIMPORT_DISCARD';
+import { parseArgs } from './generic-transformers';
+
+describe('HIMPORT DISCARD', () => {
+  describe('transformArguments', () => {
+    it('simple', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_DISCARD, 'fieldset'),
+        ['HIMPORT', 'DISCARD', 'fieldset']
+      );
+    });
+  });
+
+  describe('behavior', () => {
+    testUtils.isVersionGreaterThanHook([8, 10]);
+
+    testUtils.testAll('hImportDiscard', async client => {
+      await client.hImportPrepare('fieldset', ['f1', 'f2']);
+
+      assert.equal(
+        await client.hImportDiscard('fieldset'),
+        1
+      );
+
+      assert.equal(
+        await client.hImportDiscard('fieldset'),
+        0
+      );
+    }, {
+      client: GLOBAL.SERVERS.OPEN,
+      cluster: GLOBAL.CLUSTERS.OPEN
+    });
+  });
+});

packages/client/lib/commands/HIMPORT_DISCARD.ts

@@ -0,0 +1,9 @@
+import { CommandParser } from '../client/parser';
+import { NumberReply, Command } from '../RESP/types';
+
+export default {
+  parseCommand(parser: CommandParser, fieldset: string) {
+    parser.push('HIMPORT', 'DISCARD', fieldset);
+  },
+  transformReply: undefined as unknown as () => NumberReply
+} as const satisfies Command;

packages/client/lib/commands/HIMPORT_DISCARDALL.spec.ts

@@ -0,0 +1,36 @@
+import { strict as assert } from 'node:assert';
+import testUtils, { GLOBAL } from '../test-utils';
+import HIMPORT_DISCARDALL from './HIMPORT_DISCARDALL';
+import { parseArgs } from './generic-transformers';
+
+describe('HIMPORT DISCARDALL', () => {
+  describe('transformArguments', () => {
+    it('simple', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_DISCARDALL),
+        ['HIMPORT', 'DISCARDALL']
+      );
+    });
+  });
+
+  describe('behavior', () => {
+    testUtils.isVersionGreaterThanHook([8, 10]);
+
+    testUtils.testAll('hImportDiscardAll', async client => {
+      assert.equal(
+        await client.hImportDiscardAll(),
+        0
+      );
+
+      await client.hImportPrepare('fieldset', ['f1', 'f2']);
+
+      assert.equal(
+        await client.hImportDiscardAll(),
+        1
+      );
+    }, {
+      client: GLOBAL.SERVERS.OPEN,
+      cluster: GLOBAL.CLUSTERS.OPEN
+    });
+  });
+});

packages/client/lib/commands/HIMPORT_DISCARDALL.ts

@@ -0,0 +1,9 @@
+import { CommandParser } from '../client/parser';
+import { NumberReply, Command } from '../RESP/types';
+
+export default {
+  parseCommand(parser: CommandParser) {
+    parser.push('HIMPORT', 'DISCARDALL');
+  },
+  transformReply: undefined as unknown as () => NumberReply
+} as const satisfies Command;

packages/client/lib/commands/HIMPORT_PREPARE.spec.ts

@@ -0,0 +1,50 @@
+import { strict as assert } from 'node:assert';
+import testUtils, { GLOBAL } from '../test-utils';
+import HIMPORT_PREPARE from './HIMPORT_PREPARE';
+import { parseArgs } from './generic-transformers';
+
+describe('HIMPORT PREPARE', () => {
+  describe('transformArguments', () => {
+    it('string', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_PREPARE, 'fieldset', 'field'),
+        ['HIMPORT', 'PREPARE', 'fieldset', 'field']
+      );
+    });
+
+    it('array', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_PREPARE, 'fieldset', ['f1', 'f2']),
+        ['HIMPORT', 'PREPARE', 'fieldset', 'f1', 'f2']
+      );
+    });
+
+    it('preserves caller field order', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_PREPARE, 'fieldset', ['c', 'a', 'b']),
+        ['HIMPORT', 'PREPARE', 'fieldset', 'c', 'a', 'b']
+      );
+    });
+  });
+
+  describe('behavior', () => {
+    testUtils.isVersionGreaterThanHook([8, 10]);
+
+    testUtils.testAll('hImportPrepare', async client => {
+      assert.equal(
+        await client.hImportPrepare('fieldset', ['f1', 'f2']),
+        'OK'
+      );
+    }, {
+      client: GLOBAL.SERVERS.OPEN,
+      cluster: GLOBAL.CLUSTERS.OPEN
+    });
+
+    testUtils.testWithClient('rejects duplicate field names', async client => {
+      await assert.rejects(
+        client.hImportPrepare('fieldset', ['f1', 'f1']),
+        /duplicate field name/
+      );
+    }, GLOBAL.SERVERS.OPEN);
+  });
+});

packages/client/lib/commands/HIMPORT_PREPARE.ts

@@ -0,0 +1,11 @@
+import { CommandParser } from '../client/parser';
+import { SimpleStringReply, Command } from '../RESP/types';
+import { RedisVariadicArgument } from './generic-transformers';
+
+export default {
+  parseCommand(parser: CommandParser, fieldset: string, fields: RedisVariadicArgument) {
+    parser.push('HIMPORT', 'PREPARE', fieldset);
+    parser.pushVariadic(fields);
+  },
+  transformReply: undefined as unknown as () => SimpleStringReply<'OK'>
+} as const satisfies Command;

packages/client/lib/commands/HIMPORT_SET.spec.ts

@@ -0,0 +1,63 @@
+import { strict as assert } from 'node:assert';
+import testUtils, { GLOBAL } from '../test-utils';
+import HIMPORT_SET from './HIMPORT_SET';
+import { parseArgs } from './generic-transformers';
+
+describe('HIMPORT SET', () => {
+  describe('transformArguments', () => {
+    it('string', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_SET, 'key', 'fieldset', 'value'),
+        ['HIMPORT', 'SET', 'key', 'fieldset', 'value']
+      );
+    });
+
+    it('array', () => {
+      assert.deepEqual(
+        parseArgs(HIMPORT_SET, 'key', 'fieldset', ['v1', 'v2']),
+        ['HIMPORT', 'SET', 'key', 'fieldset', 'v1', 'v2']
+      );
+    });
+  });
+
+  describe('behavior', () => {
+    testUtils.isVersionGreaterThanHook([8, 10]);
+
+    testUtils.testAll('hImportSet roundtrip', async client => {
+      await client.hImportPrepare('fieldset', ['f1', 'f2']);
+
+      assert.equal(
+        await client.hImportSet('key', 'fieldset', ['v1', 'v2']),
+        'OK'
+      );
+
+      // enumeration order is canonicalized server-side — assert content, not order
+      assert.deepEqual(
+        await client.hGetAll('key'),
+        { f1: 'v1', f2: 'v2' }
+      );
+    }, {
+      client: GLOBAL.SERVERS.OPEN,
+      cluster: GLOBAL.CLUSTERS.OPEN
+    });
+
+    testUtils.testWithClient('rejects on value count mismatch', async client => {
+      await client.hImportPrepare('fieldset', ['f1', 'f2']);
+
+      await assert.rejects(
+        client.hImportSet('key', 'fieldset', ['v1']),
+        /value count/
+      );
+    }, GLOBAL.SERVERS.OPEN);
+
+    testUtils.testWithClient('rejects with WRONGTYPE on non-hash key', async client => {
+      await client.set('key', 'string');
+      await client.hImportPrepare('fieldset', ['f1']);
+
+      await assert.rejects(
+        client.hImportSet('key', 'fieldset', ['v1']),
+        /WRONGTYPE/
+      );
+    }, GLOBAL.SERVERS.OPEN);
+  });
+});

packages/client/lib/commands/HIMPORT_SET.ts

@@ -0,0 +1,13 @@
+import { CommandParser } from '../client/parser';
+import { RedisArgument, SimpleStringReply, Command } from '../RESP/types';
+import { RedisVariadicArgument } from './generic-transformers';
+
+export default {
+  parseCommand(parser: CommandParser, key: RedisArgument, fieldset: string, values: RedisVariadicArgument) {
+    parser.push('HIMPORT', 'SET');
+    parser.pushKey(key);
+    parser.push(fieldset);
+    parser.pushVariadic(values);
+  },
+  transformReply: undefined as unknown as () => SimpleStringReply<'OK'>
+} as const satisfies Command;