From c01a49863994aeda5b7590c4e592a829444d9001 Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Wed, 29 Jul 2026 13:25:56 +0300 Subject: [PATCH 1/3] feat(client): add HIMPORT command family with managed fieldset lifecycle Adds first-class support for the Redis 8.10 HIMPORT command family (hinted hash templates): HIMPORT PREPARE/SET/DISCARD/DISCARDALL, exposed as hImportPrepare/hImportSet/hImportDiscard/hImportDiscardAll and marked @experimental. Fieldsets are server-side session state scoped to one physical connection, so the client owns the fieldset lifecycle: a shared FieldsetRegistry records registrations on the logical client, and a per-connection hook transparently pipelines PREPARE in front of SET on sessions that lack the fieldset or hold a stale field list, replays pending discards before dependent commands, and retries a SET once on 'no such fieldset' for registered fieldsets. DISCARD/DISCARDALL replies are registry-based, giving deterministic semantics across topologies, and a rejected user discard restores the session claims and knocks the synced count back so the discard replays on that connection. On a cluster each serving node client runs the same hook, so sessions prepare and reconcile lazily per node; the eager all_shards fan-out and its response-policy pins arrive with the request-response-policies work. Co-Authored-By: Claude Fable 5 --- packages/client/lib/client/index.ts | 269 ++++++++++++++++++ packages/client/lib/client/pool.ts | 7 + packages/client/lib/cluster/cluster-slots.ts | 8 + .../lib/commands/HIMPORT_DISCARD.spec.ts | 36 +++ .../client/lib/commands/HIMPORT_DISCARD.ts | 9 + .../lib/commands/HIMPORT_DISCARDALL.spec.ts | 36 +++ .../client/lib/commands/HIMPORT_DISCARDALL.ts | 9 + .../lib/commands/HIMPORT_PREPARE.spec.ts | 50 ++++ .../client/lib/commands/HIMPORT_PREPARE.ts | 11 + .../client/lib/commands/HIMPORT_SET.spec.ts | 63 ++++ packages/client/lib/commands/HIMPORT_SET.ts | 13 + packages/client/lib/commands/index.ts | 114 ++++++++ packages/client/lib/himport/registry.spec.ts | 178 ++++++++++++ packages/client/lib/himport/registry.ts | 187 ++++++++++++ .../client/lib/himport/transparency.spec.ts | 255 +++++++++++++++++ packages/client/lib/sentinel/index.ts | 5 + 16 files changed, 1250 insertions(+) create mode 100644 packages/client/lib/commands/HIMPORT_DISCARD.spec.ts create mode 100644 packages/client/lib/commands/HIMPORT_DISCARD.ts create mode 100644 packages/client/lib/commands/HIMPORT_DISCARDALL.spec.ts create mode 100644 packages/client/lib/commands/HIMPORT_DISCARDALL.ts create mode 100644 packages/client/lib/commands/HIMPORT_PREPARE.spec.ts create mode 100644 packages/client/lib/commands/HIMPORT_PREPARE.ts create mode 100644 packages/client/lib/commands/HIMPORT_SET.spec.ts create mode 100644 packages/client/lib/commands/HIMPORT_SET.ts create mode 100644 packages/client/lib/himport/registry.spec.ts create mode 100644 packages/client/lib/himport/registry.ts create mode 100644 packages/client/lib/himport/transparency.spec.ts diff --git a/packages/client/lib/client/index.ts b/packages/client/lib/client/index.ts index 9e491c9c396..adbbc9bfb92 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -25,9 +25,35 @@ import { ClientMetricsHandle, ClientRegistry } from '../opentelemetry'; import { ClientIdentity, ClientRole, generateClientId } from './identity'; import { trace, sanitizeArgs, publish, CHANNELS, type CommandTraceContext } from './tracing'; import { DEFAULT_COMMAND_TIMEOUT } from '../defaults'; +import { FieldsetRegistry, PreparedFieldsets } from '../himport/registry'; +import HIMPORT_DISCARD from '../commands/HIMPORT_DISCARD'; +import HIMPORT_DISCARDALL from '../commands/HIMPORT_DISCARDALL'; +import HIMPORT_PREPARE from '../commands/HIMPORT_PREPARE'; +import HIMPORT_SET from '../commands/HIMPORT_SET'; const noop = () => {}; +const HIMPORT_SESSION_SUBCOMMANDS = new Set(['PREPARE', 'DISCARD', 'DISCARDALL']); + +/** + * MULTI/pipeline stores raw args only, so the HIMPORT transparency hook never sees these + * commands — a PREPARE/DISCARD executed that way would silently diverge the client registry + * from server state (a fieldset the registry doesn't know about, or a discarded one it + * would resurrect via lazy prepare). Rejected client-side at the exec funnel, which covers + * both the typed multi methods and raw `multi.addCommand(...)`. HIMPORT SET stays allowed: + * it mutates no registry state (the fieldset must already exist on the carrying connection). + */ +function assertNoHimportSessionCommands(commands: Array) { + for (const { args } of commands) { + if (String(args[0]).toUpperCase() !== 'HIMPORT') continue; + if (HIMPORT_SESSION_SUBCOMMANDS.has(String(args[1]).toUpperCase())) { + throw new Error( + 'HIMPORT PREPARE/DISCARD/DISCARDALL are not supported inside MULTI/pipeline; call them on the client before the transaction' + ); + } + } +} + export interface RedisClientOptions< M extends RedisModules = RedisModules, F extends RedisFunctions = RedisFunctions, @@ -161,6 +187,14 @@ export interface RedisClientOptions< * ``` */ clientSideCache?: ClientSideCacheProvider | ClientSideCacheConfig; + /** + * @internal + * Shared HIMPORT fieldset registry. Pool/cluster/sentinel construct one instance and inject + * it into every client they own so a fieldset registered on the logical client can be + * transparently re-prepared on any physical connection; `duplicate()` injects the parent's. + * Not a user-facing option — when omitted the client owns a fresh registry. + */ + himportRegistry?: FieldsetRegistry; /** * If set to true, disables sending client identifier (user-agent like message) to the redis server */ @@ -549,6 +583,10 @@ export default class RedisClient< #dirtyWatch?: string; #watchEpoch?: number; #clientSideCache?: ClientSideCacheProvider; + // What the user registered on the logical client (shared across pooled/cluster/sentinel + // clients via the himportRegistry option) vs. what THIS socket's server session holds. + #himportRegistry: FieldsetRegistry; + #preparedFieldsets = new PreparedFieldsets(); #credentialsSubscription: Disposable | null = null; // Flag used to pause writing to the socket during maintenance windows. // When true, prevents new commands from being written while waiting for: @@ -688,6 +726,7 @@ export default class RedisClient< this.#queue = this.#initiateQueue(this.#clientIdentity.id); this.#socket = this.#initiateSocket(this.#clientIdentity.id); + this.#himportRegistry = this.#options.himportRegistry ?? new FieldsetRegistry(); this.#registerForMetrics(); @@ -981,6 +1020,10 @@ export default class RedisClient< .on('error', err => { this.emit('error', err); this.#clientSideCache?.onError(); + // Session state died with the socket. Cleared HERE (not on re-ready): a hImportSet + // issued between disconnect and re-ready would otherwise see a live-looking entry, + // skip the PREPARE injection, and replay bare onto the new socket's empty session. + this.#preparedFieldsets.clear(); if (this.#socket.isOpen && !this.#options.disableOfflineQueue) { this.#queue.flushWaitingForReply(err); } else { @@ -1141,6 +1184,9 @@ export default class RedisClient< >(overrides?: Partial>) { return new (Object.getPrototypeOf(this).constructor)({ ...this._self.#options, + // The options spread only carries the registry if the user originally passed one — + // inject the live instance explicitly: a duplicate SHARES the parent's registrations. + himportRegistry: this._self.#himportRegistry, commandOptions: this._commandOptions, ...overrides }) as RedisClientType<_M, _F, _S, _RESP, _TYPE_MAPPING>; @@ -1175,6 +1221,9 @@ export default class RedisClient< if(this._self.#socket) { this._self._ejectSocket().destroy(); } + // A swapped-in socket carries a fresh server session with no other observable signal + // on this client — the explicit wipe is the only fieldset-invalidation mechanism here. + this._self.#preparedFieldsets.clear(); this._self.#socket = socket; this._self.#attachListeners(this._self.#socket); } @@ -1225,6 +1274,13 @@ export default class RedisClient< commandOptions: CommandOptions | undefined, transformReply: TransformReply | undefined, ) { + if ( + command === HIMPORT_SET || command === HIMPORT_PREPARE || + command === HIMPORT_DISCARD || command === HIMPORT_DISCARDALL + ) { + return this._self.#executeHimport(this, command, parser, commandOptions, transformReply); + } + const csc = this._self.#clientSideCache; const defaultTypeMapping = this._self.#options.commandOptions === commandOptions || (this._self.#options.commandOptions?.typeMapping === commandOptions?.typeMapping); @@ -1244,6 +1300,209 @@ export default class RedisClient< } } + /** + * HIMPORT transparency layer (design: ../himport/registry.ts module doc). Runs instead of + * the normal `_executeCommand` tail for the four HIMPORT commands and keeps this + * connection's server session coherent with the client-level fieldset registry: + * + * 1. Reconcile — replays DISCARDs this session missed (it was serving other commands when + * the user discarded on another connection). + * 2. Lazy prepare — pipelines a PREPARE in front of a SET when this session lacks the + * fieldset or holds a stale field-list version. No await in between: the server + * processes them in arrival order on one socket. + * 3. Registry-based replies — DISCARD/DISCARDALL resolve with the registry mutation result + * (`1`/count of registrations removed), not the per-session server reply, so the answer + * is deterministic across pools and clusters. + * 4. Retry-once — a SET failing with `no such fieldset` for a *registered* fieldset means + * this session lost its state through a path the client did not observe; re-prepare on + * this same connection and retry a single time. + * + * Raw `sendCommand(['HIMPORT', ...])` bypasses this layer by construction. + */ + async #executeHimport( + // The proxy-aware caller (`duplicate`d command-options carriers derive from the client), + // needed for `sendCommand` merging and `_commandOptions`; `this` inside is always _self. + client: RedisClient, + command: Command, + parser: CommandParser, + commandOptions: CommandOptions | undefined, + transformReply: TransformReply | undefined, + retried = false + ): Promise { + const registry = this.#himportRegistry; + const prepared = this.#preparedFieldsets; + const args = parser.redisArgs; + // The injected commands must ride with the main command's effective queue placement: + // same chainId (ASK-redirect chains flush per chain) and same asap-ness (asap unshifts + // to the queue front — a non-asap injection would let an asap SET jump its own PREPARE). + const effectiveAsap = commandOptions?.asap ?? client._commandOptions?.asap; + const effectiveChainId = commandOptions?.chainId ?? client._commandOptions?.chainId; + const injectOpts: CommandOptions = { asap: effectiveAsap, chainId: effectiveChainId }; + + // Commands to reach the wire BEFORE the main command, in this order. Each carries the + // rollback undoing its optimistic bookkeeping if the server rejects it. + const prelude: Array<{ args: Array, rollback: () => void }> = []; + + // -- Reconcile: replay discards this session may still be holding. A DISCARD must + // precede a same-name SET on the wire, or the SET would write through the discarded + // template; hence prelude, not fire-and-forget. + if (prepared.syncedDiscardCount < registry.discardCount) { + const countAtInjection = registry.discardCount; + const pending = registry.diff(prepared.names()); + // Every session name is pending → one DISCARDALL wipes the session in a single + // command. Safe precisely because nothing this session holds is worth keeping. + const collapseToDiscardAll = pending.size > 0 && pending.size === prepared.size; + const rollbackVersions = new Map(); + for (const name of pending) { + rollbackVersions.set(name, prepared.get(name)!); + prepared.delete(name); + } + prepared.syncedDiscardCount = countAtInjection; + // Failed injected DISCARD → the fieldset is still alive on this session with no + // client-side trace; restore the trace and knock the synced count back so the next + // HIMPORT command re-reconciles. Guarded so a newer PREPARE is never clobbered. + const rollback = (names: Iterable) => { + for (const name of names) { + const version = rollbackVersions.get(name); + if (version !== undefined && prepared.get(name) === undefined) { + prepared.set(name, version); + } + } + prepared.syncedDiscardCount = Math.min(prepared.syncedDiscardCount, countAtInjection - 1); + }; + if (collapseToDiscardAll) { + prelude.push({ args: ['HIMPORT', 'DISCARDALL'], rollback: () => rollback(pending) }); + } else { + for (const name of pending) { + prelude.push({ args: ['HIMPORT', 'DISCARD', name], rollback: () => rollback([name]) }); + } + } + } + + // -- Per-command registry bookkeeping, optimistic (before any reply) so concurrent + // same-tick commands see the final state and don't double-prepare (NF.2). + let userPrepare: { name: string, version: number } | undefined; + let registryReply: number | undefined; + let setName: string | undefined; + let userDiscard: { restore: Map, countAfter: number } | undefined; + + if (command === HIMPORT_PREPARE) { + const name = String(args[2]); + registry.set(name, args.slice(3)); + const version = registry.get(name)!.version; + userPrepare = { name, version }; + prepared.set(name, version); + } else if (command === HIMPORT_SET) { + // args layout: [HIMPORT, SET, key, fieldset, ...values] — index 2 is the (possibly + // keyPrefix-affected) key, index 3 is the fieldset name. + const name = String(args[3]); + setName = name; + const entry = registry.get(name); + if (entry !== undefined) { + const sessionVersion = prepared.get(name); + if (sessionVersion === undefined || sessionVersion < entry.version) { + prepared.set(name, entry.version); + prelude.push({ + args: ['HIMPORT', 'PREPARE', name, ...entry.fields], + // Injected-PREPARE failures are often transient (LOADING, failover) — roll back + // only the session claim, never the registration, or a valid fieldset would + // permanently lose lazy re-prepare. + rollback: () => { + if (prepared.get(name) === entry.version) prepared.delete(name); + } + }); + } + } + // Name absent from the registry → send as-is; the server's `no such fieldset` is + // authoritative and must not be masked. + } else if (command === HIMPORT_DISCARD) { + const name = String(args[2]); + const sessionVersion = prepared.get(name); + if (registry.discard(name)) { + registryReply = 1; + userDiscard = { + restore: sessionVersion === undefined ? new Map() : new Map([[name, sessionVersion]]), + countAfter: registry.discardCount + }; + } else { + registryReply = 0; + } + prepared.delete(name); + } else { + const restore = new Map(prepared.entries()); + registryReply = registry.discardAll(); + if (registryReply > 0) { + userDiscard = { restore, countAfter: registry.discardCount }; + } + prepared.clear(); + } + + // -- Enqueue. Everything below runs in one synchronous tick, so the prelude and the + // main command flush to the socket in a single write (the HLD-blessed pipelining). + const send = () => client.sendCommand(parser.redisArgs, commandOptions); + let mainPromise: Promise; + if (effectiveAsap) { + // asap unshifts, so consecutive front-insertions reverse: enqueue the main command + // first, then the prelude back-to-front — the wire sees prelude order, then main. + mainPromise = send(); + for (let i = prelude.length - 1; i >= 0; i--) { + const injection = prelude[i]; + client.sendCommand(injection.args, injectOpts).catch(injection.rollback); + } + } else { + for (const injection of prelude) { + client.sendCommand(injection.args, injectOpts).catch(injection.rollback); + } + mainPromise = send(); + } + + let reply: unknown; + try { + reply = await mainPromise; + } catch (err) { + if (command === HIMPORT_PREPARE && userPrepare !== undefined) { + // A rejected user PREPARE (e.g. duplicate field name) must not leave a registration + // that lazy prepare would replay forever. Version-guarded: a newer successful + // PREPARE must not be clobbered. Deleting through `discard()` also bumps + // discardCount, so sessions still holding an OLDER field list for this name + // reconcile it away instead of silently serving stale SETs. + const { name, version } = userPrepare; + if (prepared.get(name) === version) prepared.delete(name); + if (registry.get(name)?.version === version) registry.discard(name); + } else if ( + command === HIMPORT_SET && setName !== undefined && !retried && + (err as Error)?.message?.includes?.('no such fieldset') && + registry.get(setName) !== undefined + ) { + // Recover-and-retry-once (HLD NF.4): the session claim lied — state was lost through + // a path the client did not observe. Blind retries on another connection would fail + // identically; re-prepare HERE and retry a single time. + prepared.delete(setName); + return this.#executeHimport(client, command, parser, commandOptions, transformReply, true); + } else if (userDiscard !== undefined) { + // A rejected user DISCARD/DISCARDALL leaves this session's server state unknown while + // the registry mutation stands (a discard is recorded user intent — other sessions + // reconcile off the count bump regardless). Restore this session's claims and knock + // the synced count back so the next HIMPORT command replays the discard here — the + // same recovery as a failed injected DISCARD. Absence-guarded: a re-PREPARE that won + // the race keeps its entry (its name is back in the registry, so reconcile skips it). + for (const [name, version] of userDiscard.restore) { + if (prepared.get(name) === undefined) prepared.set(name, version); + } + prepared.syncedDiscardCount = Math.min(prepared.syncedDiscardCount, userDiscard.countAfter - 1); + } + throw err; + } + + const finalReply = registryReply !== undefined + ? registryReply + : transformReply ? transformReply(reply, parser.preserve, commandOptions?.typeMapping) : reply; + + publish(CHANNELS.COMMAND_REPLY, () => ({ args: sanitizeArgs(parser.redisArgs), reply: finalReply, clientId: this._clientId })); + + return finalReply; + } + /** * @internal */ @@ -1510,6 +1769,8 @@ export default class RedisClient< commands: Array, selectedDB?: number ) { + assertNoHimportSessionCommands(commands); + if (!this._self.#socket.isOpen) { return Promise.reject(new ClientClosedError()); } @@ -1566,6 +1827,8 @@ export default class RedisClient< commands: Array, selectedDB?: number ) { + assertNoHimportSessionCommands(commands); + const dirtyWatch = this._self.#dirtyWatch; this._self.#dirtyWatch = undefined; const watchEpoch = this._self.#watchEpoch; @@ -1754,6 +2017,12 @@ export default class RedisClient< * Reset the client to its default state (i.e. stop PubSub, stop monitoring, select default DB, etc.) */ async reset() { + // RESET wipes the server session's fieldsets. Cleared at entry, not after the + // round-trip: RESET queues at the #toWrite tail, so a concurrent hImportSet whose hook + // runs before a post-completion clear would see a live entry and queue a bare SET + // BEHIND the RESET — onto the wiped session. With the entry-time clear it re-injects + // PREPARE, which queues behind RESET too, in the correct order. + this._self.#preparedFieldsets.clear(); const chainId = Symbol('Reset Chain'), promises = [this._self.#queue.reset(chainId)], selectedDB = this._self.#options?.database ?? 0; diff --git a/packages/client/lib/client/pool.ts b/packages/client/lib/client/pool.ts index 34942bcca4a..dac53f71d3c 100644 --- a/packages/client/lib/client/pool.ts +++ b/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. diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index b40680531fb..461a2c0319b 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/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,12 @@ export default class RedisClusterSlots< readonly nodeByAddress = new Map | ShardNode>(); pubSubNode?: PubSubNode; 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 = new FieldsetRegistry(); smigratedSeqIdsSeen = new Set; #topologyRefreshPromise?: Promise; @@ -603,6 +610,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, diff --git a/packages/client/lib/commands/HIMPORT_DISCARD.spec.ts b/packages/client/lib/commands/HIMPORT_DISCARD.spec.ts new file mode 100644 index 00000000000..a71145a163e --- /dev/null +++ b/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 + }); + }); +}); diff --git a/packages/client/lib/commands/HIMPORT_DISCARD.ts b/packages/client/lib/commands/HIMPORT_DISCARD.ts new file mode 100644 index 00000000000..3559c9e9e37 --- /dev/null +++ b/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; diff --git a/packages/client/lib/commands/HIMPORT_DISCARDALL.spec.ts b/packages/client/lib/commands/HIMPORT_DISCARDALL.spec.ts new file mode 100644 index 00000000000..56d9455a0f5 --- /dev/null +++ b/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 + }); + }); +}); diff --git a/packages/client/lib/commands/HIMPORT_DISCARDALL.ts b/packages/client/lib/commands/HIMPORT_DISCARDALL.ts new file mode 100644 index 00000000000..fa658a57368 --- /dev/null +++ b/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; diff --git a/packages/client/lib/commands/HIMPORT_PREPARE.spec.ts b/packages/client/lib/commands/HIMPORT_PREPARE.spec.ts new file mode 100644 index 00000000000..1d19143b191 --- /dev/null +++ b/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); + }); +}); diff --git a/packages/client/lib/commands/HIMPORT_PREPARE.ts b/packages/client/lib/commands/HIMPORT_PREPARE.ts new file mode 100644 index 00000000000..babfefd584f --- /dev/null +++ b/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; diff --git a/packages/client/lib/commands/HIMPORT_SET.spec.ts b/packages/client/lib/commands/HIMPORT_SET.spec.ts new file mode 100644 index 00000000000..20b5d7d5dd6 --- /dev/null +++ b/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); + }); +}); diff --git a/packages/client/lib/commands/HIMPORT_SET.ts b/packages/client/lib/commands/HIMPORT_SET.ts new file mode 100644 index 00000000000..b0c37eae62a --- /dev/null +++ b/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; diff --git a/packages/client/lib/commands/index.ts b/packages/client/lib/commands/index.ts index 65b98a70f61..96f76c6bf82 100644 --- a/packages/client/lib/commands/index.ts +++ b/packages/client/lib/commands/index.ts @@ -163,6 +163,10 @@ import HGET from './HGET'; import HGETALL from './HGETALL'; import HGETDEL from './HGETDEL'; import HGETEX from './HGETEX'; +import HIMPORT_DISCARD from './HIMPORT_DISCARD'; +import HIMPORT_DISCARDALL from './HIMPORT_DISCARDALL'; +import HIMPORT_PREPARE from './HIMPORT_PREPARE'; +import HIMPORT_SET from './HIMPORT_SET'; import HINCRBY from './HINCRBY'; import HINCRBYFLOAT from './HINCRBYFLOAT'; import HKEYS from './HKEYS'; @@ -2373,6 +2377,116 @@ export default { * @param options - Options for setting expiration */ hGetEx: HGETEX, + /** + * + * @experimental + * + * Removes a fieldset registration from the client and discards it from the server + * connection session. The reply is registry-based: `1` if the fieldset was registered on + * this client and removed, `0` otherwise. + * @param fieldset - Name of the fieldset to discard + * @see https://redis.io/commands/himport-discard/ + * @since 8.10 + */ + HIMPORT_DISCARD, + /** + * + * @experimental + * + * Removes a fieldset registration from the client and discards it from the server + * connection session. The reply is registry-based: `1` if the fieldset was registered on + * this client and removed, `0` otherwise. + * @param fieldset - Name of the fieldset to discard + * @see https://redis.io/commands/himport-discard/ + * @since 8.10 + */ + hImportDiscard: HIMPORT_DISCARD, + /** + * + * @experimental + * + * Removes all fieldset registrations from the client and discards them from the server + * connection session. The reply is registry-based: the number of fieldsets that were + * registered on this client and removed. + * @see https://redis.io/commands/himport-discardall/ + * @since 8.10 + */ + HIMPORT_DISCARDALL, + /** + * + * @experimental + * + * Removes all fieldset registrations from the client and discards them from the server + * connection session. The reply is registry-based: the number of fieldsets that were + * registered on this client and removed. + * @see https://redis.io/commands/himport-discardall/ + * @since 8.10 + */ + hImportDiscardAll: HIMPORT_DISCARDALL, + /** + * + * @experimental + * + * Registers an ordered list of field names under a fieldset name for use by subsequent + * HIMPORT SET calls. The client keeps the registration and transparently re-prepares + * connections as needed (reconnects, pool growth, cluster topology changes). Raw + * `sendCommand(['HIMPORT', ...])` and raw `sendCommand(['RESET'])` bypass this managed + * layer. `duplicate()` shares the parent's registrations; `createPool()` does not. + * Inside MULTI, PREPARE/DISCARD/DISCARDALL are rejected client-side; HIMPORT SET is + * allowed if the fieldset was prepared on that connection beforehand. + * @param fieldset - Name to register the field list under + * @param fields - One or more hash field names; values in later HIMPORT SET calls map + * positionally to this order + * @see https://redis.io/commands/himport-prepare/ + * @since 8.10 + */ + HIMPORT_PREPARE, + /** + * + * @experimental + * + * Registers an ordered list of field names under a fieldset name for use by subsequent + * HIMPORT SET calls. The client keeps the registration and transparently re-prepares + * connections as needed (reconnects, pool growth, cluster topology changes). Raw + * `sendCommand(['HIMPORT', ...])` and raw `sendCommand(['RESET'])` bypass this managed + * layer. `duplicate()` shares the parent's registrations; `createPool()` does not. + * Inside MULTI, PREPARE/DISCARD/DISCARDALL are rejected client-side; HIMPORT SET is + * allowed if the fieldset was prepared on that connection beforehand. + * @param fieldset - Name to register the field list under + * @param fields - One or more hash field names; values in later HIMPORT SET calls map + * positionally to this order + * @see https://redis.io/commands/himport-prepare/ + * @since 8.10 + */ + hImportPrepare: HIMPORT_PREPARE, + /** + * + * @experimental + * + * Creates or fully replaces the hash at key using the field list of a prepared fieldset. + * Values map positionally to the fields given to HIMPORT PREPARE for that fieldset. Hash + * field enumeration order is not guaranteed to match the prepare order. + * @param key - Hash key to create or replace + * @param fieldset - Name of a fieldset registered with HIMPORT PREPARE + * @param values - Values, one per prepared field, in the prepared order + * @see https://redis.io/commands/himport-set/ + * @since 8.10 + */ + HIMPORT_SET, + /** + * + * @experimental + * + * Creates or fully replaces the hash at key using the field list of a prepared fieldset. + * Values map positionally to the fields given to HIMPORT PREPARE for that fieldset. Hash + * field enumeration order is not guaranteed to match the prepare order. + * @param key - Hash key to create or replace + * @param fieldset - Name of a fieldset registered with HIMPORT PREPARE + * @param values - Values, one per prepared field, in the prepared order + * @see https://redis.io/commands/himport-set/ + * @since 8.10 + */ + hImportSet: HIMPORT_SET, /** * Increments the integer value of a field in a hash by the given number * @param key - Key of the hash diff --git a/packages/client/lib/himport/registry.spec.ts b/packages/client/lib/himport/registry.spec.ts new file mode 100644 index 00000000000..3fc659e2a02 --- /dev/null +++ b/packages/client/lib/himport/registry.spec.ts @@ -0,0 +1,178 @@ +import { strict as assert } from 'node:assert'; +import { FieldsetRegistry, PreparedFieldsets } from './registry'; + +describe('FieldsetRegistry', () => { + describe('set', () => { + it('is idempotent: same fields keep the same version', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a', 'b']); + const version = registry.get('fs')!.version; + + registry.set('fs', ['a', 'b']); + assert.equal(registry.get('fs')!.version, version); + }); + + it('bumps the version on a changed field list', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a', 'b']); + const version = registry.get('fs')!.version; + + registry.set('fs', ['a', 'c']); + assert.ok(registry.get('fs')!.version > version); + }); + + it('field order matters — reordered fields are a new version', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a', 'b']); + const version = registry.get('fs')!.version; + + registry.set('fs', ['b', 'a']); + assert.ok(registry.get('fs')!.version > version); + }); + + it('versions stay monotonic across discard + re-prepare', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a', 'b']); + const version = registry.get('fs')!.version; + + registry.discard('fs'); + registry.set('fs', ['a', 'b']); + assert.ok(registry.get('fs')!.version > version); + }); + + it('compares fields byte-wise: Buffer and string with identical bytes are equal', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a', Buffer.from('b')]); + const version = registry.get('fs')!.version; + + registry.set('fs', [Buffer.from('a'), 'b']); + assert.equal(registry.get('fs')!.version, version); + }); + + it('stores a copy — mutating the caller array does not affect the registry', () => { + const registry = new FieldsetRegistry(); + const fields = ['a', 'b']; + registry.set('fs', fields); + + fields[1] = 'mutated'; + assert.deepEqual(registry.get('fs')!.fields, ['a', 'b']); + }); + + it('accepts empty string as a fieldset name', () => { + const registry = new FieldsetRegistry(); + registry.set('', ['a']); + assert.ok(registry.get('') !== undefined); + }); + }); + + describe('discard', () => { + it('returns true and bumps discardCount for a registered name', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a']); + + assert.equal(registry.discard('fs'), true); + assert.equal(registry.discardCount, 1); + assert.equal(registry.get('fs'), undefined); + }); + + it('no-op discard returns false and does not bump discardCount', () => { + const registry = new FieldsetRegistry(); + + assert.equal(registry.discard('missing'), false); + assert.equal(registry.discardCount, 0); + }); + + it('repeated discards bump discardCount once', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a']); + + registry.discard('fs'); + registry.discard('fs'); + registry.discard('fs'); + assert.equal(registry.discardCount, 1); + }); + }); + + describe('discardAll', () => { + it('returns the number of removed registrations with a single discardCount bump', () => { + const registry = new FieldsetRegistry(); + registry.set('fs1', ['a']); + registry.set('fs2', ['b']); + + assert.equal(registry.discardAll(), 2); + assert.equal(registry.discardCount, 1); + assert.equal(registry.get('fs1'), undefined); + }); + + it('no-op on an empty registry: returns 0, no bump', () => { + const registry = new FieldsetRegistry(); + + assert.equal(registry.discardAll(), 0); + assert.equal(registry.discardCount, 0); + }); + }); + + describe('diff', () => { + it('returns exactly the session-only names', () => { + const registry = new FieldsetRegistry(); + registry.set('kept', ['a']); + registry.set('discarded', ['b']); + registry.discard('discarded'); + + assert.deepEqual( + registry.diff(new Set(['kept', 'discarded'])), + new Set(['discarded']) + ); + }); + + it('a discarded and re-registered name is not pending', () => { + const registry = new FieldsetRegistry(); + registry.set('fs', ['a']); + registry.discard('fs'); + registry.set('fs', ['b']); + + assert.deepEqual( + registry.diff(new Set(['fs'])), + new Set() + ); + }); + }); +}); + +describe('PreparedFieldsets', () => { + it('tracks name → version', () => { + const prepared = new PreparedFieldsets(); + prepared.set('fs', 3); + + assert.equal(prepared.get('fs'), 3); + assert.equal(prepared.get('missing'), undefined); + }); + + it('names() returns a snapshot of tracked names', () => { + const prepared = new PreparedFieldsets(); + prepared.set('fs1', 1); + prepared.set('fs2', 2); + + assert.deepEqual(prepared.names(), new Set(['fs1', 'fs2'])); + }); + + it('clear() wipes entries but keeps syncedDiscardCount', () => { + const prepared = new PreparedFieldsets(); + prepared.set('fs', 1); + prepared.syncedDiscardCount = 5; + + prepared.clear(); + assert.equal(prepared.size, 0); + assert.equal(prepared.syncedDiscardCount, 5); + }); + + it('delete() removes a single entry', () => { + const prepared = new PreparedFieldsets(); + prepared.set('fs1', 1); + prepared.set('fs2', 2); + + assert.equal(prepared.delete('fs1'), true); + assert.equal(prepared.delete('fs1'), false); + assert.deepEqual(prepared.names(), new Set(['fs2'])); + }); +}); diff --git a/packages/client/lib/himport/registry.ts b/packages/client/lib/himport/registry.ts new file mode 100644 index 00000000000..39fea4bc8c1 --- /dev/null +++ b/packages/client/lib/himport/registry.ts @@ -0,0 +1,187 @@ +import { RedisArgument } from '../RESP/types'; + +/** + * Client-side bookkeeping for the HIMPORT command family (hinted hash templates, Redis 8.10). + * + * Server-side fieldsets are session state attached to one physical connection: they die on + * disconnect and RESET, and no other connection can see them. The client therefore keeps two + * layers of state: + * + * - {@link FieldsetRegistry} — one per logical client (standalone client, pool, cluster, + * sentinel), shared by every `RedisClient` the logical client owns. It records what the + * user *registered* (`hImportPrepare`) and is the replay source for transparently + * re-preparing connections. + * - {@link PreparedFieldsets} — one per `RedisClient`, never shared. It mirrors what that + * client's *current socket session* actually holds, so the lazy-prepare hook can decide + * whether an `HIMPORT SET` needs a `PREPARE` pipelined in front of it. + * + * TODO(himport-multi) — auto-prepare inside MULTI/pipeline. The multi queue stores raw args + * only (`multi-command.ts`) — no `Command` identity survives. Recipe when picked up: in + * `_executeMulti` sniff HIMPORT SET raw args (Buffer-tolerant compare of args[0]/args[1]) and + * inject PREPAREs under the same `chainId` BEFORE the `['MULTI']` (fieldsets are + * session-level — the PREPARE need not be inside the transaction); keep injected promises out + * of the positional result mapping (`transformReplies` uses only the last reply); same + * treatment in `_executePipeline`. Needs aborted-EXEC registry bookkeeping thought through. + * Until then: inside MULTI the fieldset must already exist on that connection (prepare + * beforehand outside the MULTI) — documented. The session-subcommand reject guard already + * scans queued commands at the same funnel; when this TODO is picked up, revisit whether that + * reject should become a registry-mirroring sniff instead. + */ + +export interface Fieldset { + /** + * The field names exactly as the caller passed them to `hImportPrepare` — never deduped, + * sorted, or reordered (the server pairs SET values to this order positionally). + */ + fields: Array; + /** + * Staleness token from the registry-wide monotonic counter. A connection whose + * `PreparedFieldsets` entry holds a lower number must re-PREPARE before its next SET. + */ + version: number; +} + +/** + * What the user registered on the logical client: fieldset name → ordered field list. + * Single source of truth for lazy re-prepare; also the source of the registry-based + * DISCARD/DISCARDALL replies. Mutated only by the per-connection command hook. + */ +export class FieldsetRegistry { + /** Fieldset name → registration, keyed by the `String()`-coerced name. */ + #fieldsets = new Map(); + + /** Source of `Fieldset.version` — registry-wide and never reset. */ + #versionCounter = 0; + + /** + * Counts effective discards only (discarding an unknown name does not bump). Connections + * compare their `syncedDiscardCount` snapshot against this to detect pending discards. + */ + discardCount = 0; + + /** + * Idempotent upsert: registering the same name with a deep-equal field list keeps the + * existing version (cluster fan-out and per-worker startup prepares must not churn + * versions); a new name or a changed field list gets the next counter value. + */ + set(name: string, fields: Array): void { + const key = String(name); + const existing = this.#fieldsets.get(key); + if (existing !== undefined && fieldsEqual(existing.fields, fields)) return; + this.#fieldsets.set(key, { + fields: fields.slice(), + version: ++this.#versionCounter + }); + } + + get(name: string): Fieldset | undefined { + return this.#fieldsets.get(String(name)); + } + + /** + * Returns `true` iff the name was registered — this boolean IS the user-facing + * `hImportDiscard` reply (registry-based, not the server session's). + */ + discard(name: string): boolean { + const removed = this.#fieldsets.delete(String(name)); + if (removed) this.discardCount++; + return removed; + } + + /** + * Returns the number of registrations removed — this count IS the user-facing + * `hImportDiscardAll` reply. + */ + discardAll(): number { + const removed = this.#fieldsets.size; + if (removed > 0) { + this.#fieldsets.clear(); + this.discardCount++; + } + return removed; + } + + /** + * Names the session still holds but the registry no longer does — the discards that + * connection has yet to replay. + */ + diff(sessionNames: Set): Set { + const pending = new Set(); + for (const name of sessionNames) { + if (!this.#fieldsets.has(name)) pending.add(name); + } + return pending; + } +} + +/** + * What one `RedisClient`'s current socket session holds. Every entry is a CLAIM about + * server-side session state, not client state: it is tied to the socket's lifetime, so the + * client wipes this whole object on socket error, `reset()`, and socket replacement. A claim + * that survives an unenumerated loss path lies — the `no such fieldset` recover-and-retry + * net heals that at the cost of one extra round trip. + */ +export class PreparedFieldsets { + /** + * Fieldset name → the {@link Fieldset.version} this session was prepared with. A missing + * entry or a lower version means the next dependent command must pipeline a PREPARE first. + */ + #versions = new Map(); + + /** + * Snapshot of `FieldsetRegistry.discardCount` this connection has reconciled up to. + * `syncedDiscardCount < registry.discardCount` means discards happened that this session + * may still hold — reconcile before the next HIMPORT command. Knocked back when an + * injected DISCARD fails, so the reconcile re-runs. + */ + syncedDiscardCount = 0; + + get(name: string): number | undefined { + return this.#versions.get(name); + } + + set(name: string, version: number): void { + this.#versions.set(name, version); + } + + delete(name: string): boolean { + return this.#versions.delete(name); + } + + names(): Set { + return new Set(this.#versions.keys()); + } + + /** Live view over the session claims — snapshot (`new Map(entries())`) before mutating. */ + entries(): IterableIterator<[string, number]> { + return this.#versions.entries(); + } + + get size(): number { + return this.#versions.size; + } + + clear(): void { + this.#versions.clear(); + } +} + +/** + * Element-wise, byte-level equality: a `Buffer` and a `string` with identical bytes are + * equal. Byte-level matters because the idempotency decision must match what the wire would + * carry, not the JS representation. + */ +function fieldsEqual(a: Array, b: Array): boolean { + if (a.length !== b.length) return false; + for (let i = 0; i < a.length; i++) { + const x = a[i], y = b[i]; + if (typeof x === 'string' && typeof y === 'string') { + if (x !== y) return false; + } else { + const xb = typeof x === 'string' ? Buffer.from(x) : x; + const yb = typeof y === 'string' ? Buffer.from(y) : y; + if (!xb.equals(yb)) return false; + } + } + return true; +} diff --git a/packages/client/lib/himport/transparency.spec.ts b/packages/client/lib/himport/transparency.spec.ts new file mode 100644 index 00000000000..99be9f2727e --- /dev/null +++ b/packages/client/lib/himport/transparency.spec.ts @@ -0,0 +1,255 @@ +import { strict as assert } from 'node:assert'; +import { once } from 'node:events'; +import testUtils, { GLOBAL } from '../test-utils'; + +/** + * Integration tests for the HIMPORT transparency layer (`#executeHimport` + + * `FieldsetRegistry`/`PreparedFieldsets`): lazy prepare, lazy discard reconcile, + * registry-based replies, and the recover-and-retry-once net. + */ +describe('HIMPORT transparency layer', () => { + testUtils.isVersionGreaterThanHook([8, 10]); + + // Force a disconnect and wait for the socket error — the next command reconnects. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + async function killClient(client: any): Promise { + const onceErrorPromise = once(client, 'error'); + await client.sendCommand(['QUIT']); + await Promise.all([ + onceErrorPromise, + assert.rejects(client.ping()) + ]); + } + + testUtils.testWithClient('lazily re-prepares after a reconnect', async client => { + await client.hImportPrepare('fs', ['f1', 'f2']); + await killClient(client); + + // The new session has no fieldsets; the hook must pipeline a PREPARE in front. + assert.equal(await client.hImportSet('key', 'fs', ['v1', 'v2']), 'OK'); + assert.deepEqual(await client.hGetAll('key'), { f1: 'v1', f2: 'v2' }); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('re-prepares when the field list changes (version staleness)', async client => { + await client.hImportPrepare('fs', ['old1', 'old2']); + await client.hImportSet('key1', 'fs', ['v1', 'v2']); + + await client.hImportPrepare('fs', ['new1', 'new2']); + await client.hImportSet('key2', 'fs', ['v1', 'v2']); + + assert.deepEqual(await client.hGetAll('key2'), { new1: 'v1', new2: 'v2' }); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('lazily re-prepares after a typed reset()', async client => { + await client.hImportPrepare('fs', ['f1']); + await client.reset(); + + assert.equal(await client.hImportSet('key', 'fs', ['v1']), 'OK'); + assert.deepEqual(await client.hGetAll('key'), { f1: 'v1' }); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('discard stops lazy prepare; the server error propagates', async client => { + await client.hImportPrepare('fs', ['f1']); + assert.equal(await client.hImportDiscard('fs'), 1); + + await assert.rejects( + client.hImportSet('key', 'fs', ['v1']), + /no such fieldset/ + ); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('asap SET reaches the wire behind its injected PREPARE', async client => { + await client.hImportPrepare('fs', ['f1']); + // Fresh socket, empty session — the injection must unshift IN FRONT of the asap SET; + // same-order enqueue would put the SET first and fail with `no such fieldset`. + await killClient(client); + + assert.equal(await client.asap().hImportSet('key', 'fs', ['v1']), 'OK'); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('rejected user PREPARE removes the registration', async client => { + await assert.rejects( + client.hImportPrepare('fs', ['dup', 'dup']), + /duplicate field name/ + ); + + // No lazy-prepare retry loop: the registration is gone, the server error propagates. + await assert.rejects( + client.hImportSet('key', 'fs', ['v1', 'v2']), + /no such fieldset/ + ); + + // A corrected PREPARE registers fresh. + await client.hImportPrepare('fs', ['f1', 'f2']); + assert.equal(await client.hImportSet('key', 'fs', ['v1', 'v2']), 'OK'); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('retries once when the session lost its state unobserved', async client => { + await client.hImportPrepare('fs', ['f1']); + await client.hImportSet('key1', 'fs', ['v1']); + + // Raw RESET bypasses the typed reset() wipe: the server session is gone but the + // client still claims the fieldset is prepared — exactly the lie the + // recover-and-retry-once net exists to absorb. + await client.sendCommand(['RESET']); + + assert.equal(await client.hImportSet('key2', 'fs', ['v1']), 'OK'); + // hGet, not hGetAll: raw RESET also reverted the connection protocol to RESP2, so + // map-shaped replies would decode flat; a bulk string is identical in both protocols. + assert.equal(await client.hGet('key2', 'f1'), 'v1'); + }, GLOBAL.SERVERS.OPEN); + + describe('user discard failure recovery', () => { + // Reject the first matching HIMPORT wire command instead of sending it — simulates a + // server-side failure while the connection (and its session fieldsets) stays alive. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + function failNextWireCommand(client: any, subcommand: string): void { + // The hook sends through `_self`, not the `Object.create` proxy the test holds. + const self = client._self; + const original = self.sendCommand.bind(self); + self.sendCommand = (args: Array, opts?: unknown) => { + if (String(args[0]) === 'HIMPORT' && String(args[1]) === subcommand) { + self.sendCommand = original; + return Promise.reject(new Error('simulated HIMPORT failure')); + } + return original(args, opts); + }; + } + + testUtils.testWithClient('failed user DISCARD is replayed before the next dependent command', async client => { + await client.hImportPrepare('fs', ['f1']); + assert.equal(await client.hImportSet('key1', 'fs', ['v1']), 'OK'); + + failNextWireCommand(client, 'DISCARD'); + await assert.rejects(client.hImportDiscard('fs'), /simulated HIMPORT failure/); + + // The session still holds the fieldset server-side; the rollback must leave a trace + // that makes the reconcile replay the discard before this SET reaches the wire — a + // success here would be a silent write through a discarded template. + await assert.rejects( + client.hImportSet('key2', 'fs', ['v1']), + /no such fieldset/ + ); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('failed user DISCARDALL is replayed before the next dependent command', async client => { + await client.hImportPrepare('fs1', ['f1']); + await client.hImportPrepare('fs2', ['f2']); + assert.equal(await client.hImportSet('key1', 'fs1', ['v1']), 'OK'); + + failNextWireCommand(client, 'DISCARDALL'); + await assert.rejects(client.hImportDiscardAll(), /simulated HIMPORT failure/); + + // Both sessions claims were wiped optimistically; the rollback restores them so the + // reconcile (collapsed back to one DISCARDALL) wipes the server session for real. + await assert.rejects(client.hImportSet('key2', 'fs1', ['v1']), /no such fieldset/); + await assert.rejects(client.hImportSet('key3', 'fs2', ['v1']), /no such fieldset/); + }, GLOBAL.SERVERS.OPEN); + }); + + describe('MULTI/pipeline guard', () => { + testUtils.testWithClient('rejects session subcommands inside MULTI', async client => { + for (const build of [ + () => client.multi().hImportPrepare('fs', ['f1']), + () => client.multi().hImportDiscard('fs'), + () => client.multi().hImportDiscardAll(), + () => client.multi().addCommand(['HIMPORT', 'PREPARE', 'fs', 'f1']) + ]) { + await assert.rejects( + build().exec(), + /not supported inside MULTI\/pipeline/ + ); + } + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('allows hImportSet inside MULTI after an out-of-band prepare', async client => { + await client.hImportPrepare('fs', ['f1']); + // Same connection: prepare above already prepared this session. + const replies = await client.multi() + .hImportSet('key', 'fs', ['v1']) + .exec(); + + assert.deepEqual(replies, ['OK']); + assert.deepEqual(await client.hGetAll('key'), { f1: 'v1' }); + }, GLOBAL.SERVERS.OPEN); + }); + + describe('pool', () => { + testUtils.testWithClientPool('each borrowed connection lazily self-prepares', async pool => { + await pool.hImportPrepare('fs', ['f1', 'f2']); + + // Concurrent SETs spread across pooled connections; each one must transparently + // prepare its own session on first use. + const keys = ['key1', 'key2', 'key3', 'key4', 'key5', 'key6']; + const replies = await Promise.all( + keys.map(key => pool.hImportSet(key, 'fs', ['v1', 'v2'])) + ); + assert.deepEqual(replies, keys.map(() => 'OK')); + + assert.deepEqual(await pool.hGetAll('key6'), { f1: 'v1', f2: 'v2' }); + }, { + ...GLOBAL.SERVERS.OPEN, + poolOptions: { minimum: 3 } + }); + + testUtils.testWithClientPool('re-prepare with changed fields reaches every connection', async pool => { + await pool.hImportPrepare('fs', ['old1', 'old2']); + await Promise.all( + ['a1', 'a2', 'a3', 'a4'].map(key => pool.hImportSet(key, 'fs', ['v1', 'v2'])) + ); + + await pool.hImportPrepare('fs', ['new1', 'new2']); + const keys = ['b1', 'b2', 'b3', 'b4', 'b5', 'b6']; + await Promise.all(keys.map(key => pool.hImportSet(key, 'fs', ['v1', 'v2']))); + + for (const key of keys) { + assert.deepEqual(await pool.hGetAll(key), { new1: 'v1', new2: 'v2' }); + } + }, { + ...GLOBAL.SERVERS.OPEN, + poolOptions: { minimum: 3 } + }); + + testUtils.testWithClientPool('discard is coherent across connections (lazy reconcile)', async pool => { + await pool.hImportPrepare('fs', ['f1', 'f2']); + // Spread lazy prepares across connections so several sessions hold the fieldset. + await Promise.all( + ['a1', 'a2', 'a3', 'a4', 'a5', 'a6'].map(key => pool.hImportSet(key, 'fs', ['v1', 'v2'])) + ); + + // Registry-based reply: 1 regardless of which connection the DISCARD borrows. + assert.equal(await pool.hImportDiscard('fs'), 1); + assert.equal(await pool.hImportDiscard('fs'), 0); + + // No stale-SET success on ANY connection: sessions still holding the fieldset must + // reconcile the discard before the SET reaches the wire. + for (const key of ['c1', 'c2', 'c3', 'c4', 'c5', 'c6']) { + await assert.rejects( + pool.hImportSet(key, 'fs', ['v1', 'v2']), + /no such fieldset/ + ); + } + }, { + ...GLOBAL.SERVERS.OPEN, + poolOptions: { minimum: 3 } + }); + }); + + describe('cluster', () => { + testUtils.testWithCluster('SETs route by slot with per-node lazy prepare; discardAll is registry-based', async cluster => { + assert.equal(await cluster.hImportPrepare('fs', ['f1', 'f2']), 'OK'); + + // Keys hashing to different slots — each SET routes to its own master. + const keys = ['key:{1}', 'key:{2}', 'key:{3}', 'key:{4}', 'key:{5}']; + const replies = await Promise.all( + keys.map(key => cluster.hImportSet(key, 'fs', ['v1', 'v2'])) + ); + assert.deepEqual(replies, keys.map(() => 'OK')); + assert.deepEqual(await cluster.hGetAll('key:{3}'), { f1: 'v1', f2: 'v2' }); + + // Registry-based reply: exactly 1 registered fieldset removed — not a per-master + // sum, not a session count. + assert.equal(await cluster.hImportDiscardAll(), 1); + }, GLOBAL.CLUSTERS.OPEN); + }); +}); diff --git a/packages/client/lib/sentinel/index.ts b/packages/client/lib/sentinel/index.ts index 92e89f838d9..4bf145bc880 100644 --- a/packages/client/lib/sentinel/index.ts +++ b/packages/client/lib/sentinel/index.ts @@ -20,6 +20,7 @@ import { TcpNetConnectOpts } from 'node:net'; import { RedisTcpSocketOptions } from '../client/socket'; import { BasicPooledClientSideCache, PooledClientSideCacheProvider } from '../client/cache'; import { ClientIdentity, ClientRole, generateClientId } from '../client/identity'; +import { FieldsetRegistry } from '../himport/registry'; import { DEFAULT_COMMAND_TIMEOUT } from '../defaults'; import { ScanOptions } from '../commands/SCAN'; @@ -862,6 +863,10 @@ export class RedisSentinelInternal< if (this.#nodeClientOptions.url !== undefined) { throw new Error("invalid nodeClientOptions for Sentinel"); } + // One fieldset registry across master/replica node clients: fieldsets registered before + // a failover must be transparently re-preparable on the promoted master's connections. + // (Sentinel-monitor clients use #sentinelClientOptions and never run HIMPORT.) + this.#nodeClientOptions.himportRegistry = new FieldsetRegistry(); if (options.clientSideCache) { if (options.clientSideCache instanceof PooledClientSideCacheProvider) { From 492bbda127929ed516f25eef697705bb2697e754 Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Wed, 29 Jul 2026 16:40:44 +0300 Subject: [PATCH 2/3] fix(client): harden HIMPORT registry rollback across duplicate/pool/cluster Address three registry-lifecycle bugs in the HIMPORT transparency layer that broke the duplicate/pool/cluster registration-sharing contract: - Rejected replacement PREPARE dropped a still-valid registration. Snapshot the prior field list and restore it on reject; a rejected fresh PREPARE still removes the name. - RedisCluster.duplicate() lost all registrations because RedisClusterSlots always allocated a fresh registry. Thread the parent registry through an internal himportRegistry option, mirroring the standalone duplicate guarantee. - A DISCARD/DISCARDALL rejected before reaching the wire (client closed/offline, aborted signal, full queue) removed the fieldset from the shared registry even though the server never saw it. Re-register on pre-enqueue failure; wire/server failures keep the existing replay-on-next-command semantics. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/lib/client/index.ts | 78 ++++++++++++++----- packages/client/lib/cluster/cluster-slots.ts | 8 +- packages/client/lib/cluster/index.ts | 12 +++ packages/client/lib/himport/registry.spec.ts | 17 ++++ packages/client/lib/himport/registry.ts | 10 +++ .../client/lib/himport/transparency.spec.ts | 52 +++++++++++++ 6 files changed, 158 insertions(+), 19 deletions(-) diff --git a/packages/client/lib/client/index.ts b/packages/client/lib/client/index.ts index adbbc9bfb92..648c2a9fb01 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -4,7 +4,7 @@ import { BasicAuth, CredentialsError, CredentialsProvider, StreamingCredentialsP import RedisCommandsQueue, { CommandOptions } from './commands-queue'; import { EventEmitter } from 'node:events'; import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumentsPrefix } from '../commander'; -import { ClientClosedError, ClientOfflineError, DisconnectsClientError, WatchError } from '../errors'; +import { AbortError, ClientClosedError, ClientOfflineError, DisconnectsClientError, WatchError } from '../errors'; import { URL } from 'node:url'; import { TcpSocketConnectOpts } from 'node:net'; import { PUBSUB_TYPE, PubSubType, PubSubListener, PubSubTypeListeners, ChannelListeners } from './pub-sub'; @@ -332,6 +332,20 @@ export interface ScanIteratorOptions { export type MonitorCallback = (reply: ReplyWithTypeMapping) => unknown; +/** + * A `sendCommand` rejection that happened BEFORE the command reached the wire (client closed + * or offline, an already-aborted signal, or a full queue — see `sendCommand`/`addCommand`). + * The HIMPORT layer rolls back optimistic registry mutations on these because the server never + * saw the command; genuine wire/server errors (`ErrorReply`, connection drops) are left to the + * replay path instead. + */ +function isPreEnqueueError(err: unknown): boolean { + return err instanceof ClientClosedError || + err instanceof ClientOfflineError || + err instanceof AbortError || + (err instanceof Error && err.message === 'The queue is full'); +} + export default class RedisClient< M extends RedisModules, F extends RedisFunctions, @@ -1381,16 +1395,20 @@ export default class RedisClient< // -- Per-command registry bookkeeping, optimistic (before any reply) so concurrent // same-tick commands see the final state and don't double-prepare (NF.2). - let userPrepare: { name: string, version: number } | undefined; + let userPrepare: { name: string, version: number, priorFields: Array | undefined } | undefined; let registryReply: number | undefined; let setName: string | undefined; - let userDiscard: { restore: Map, countAfter: number } | undefined; + let userDiscard: { restore: Map, removed: Map>, countAfter: number } | undefined; if (command === HIMPORT_PREPARE) { const name = String(args[2]); + // Snapshot the prior registration BEFORE overwriting it: a rejected *replacement* + // PREPARE must restore it (the server keeps the old fieldset on reject), not drop the + // name and strand later SETs with `no such fieldset`. + const priorFields = registry.get(name)?.fields; registry.set(name, args.slice(3)); const version = registry.get(name)!.version; - userPrepare = { name, version }; + userPrepare = { name, version, priorFields }; prepared.set(name, version); } else if (command === HIMPORT_SET) { // args layout: [HIMPORT, SET, key, fieldset, ...values] — index 2 is the (possibly @@ -1418,10 +1436,13 @@ export default class RedisClient< } else if (command === HIMPORT_DISCARD) { const name = String(args[2]); const sessionVersion = prepared.get(name); + // Capture the removed field list so a pre-enqueue failure can re-register it. + const priorFields = registry.get(name)?.fields; if (registry.discard(name)) { registryReply = 1; userDiscard = { restore: sessionVersion === undefined ? new Map() : new Map([[name, sessionVersion]]), + removed: new Map([[name, priorFields!]]), countAfter: registry.discardCount }; } else { @@ -1430,9 +1451,11 @@ export default class RedisClient< prepared.delete(name); } else { const restore = new Map(prepared.entries()); + // Snapshot all registrations so a pre-enqueue failure can re-register them. + const removed = registry.snapshot(); registryReply = registry.discardAll(); if (registryReply > 0) { - userDiscard = { restore, countAfter: registry.discardCount }; + userDiscard = { restore, removed, countAfter: registry.discardCount }; } prepared.clear(); } @@ -1461,14 +1484,23 @@ export default class RedisClient< reply = await mainPromise; } catch (err) { if (command === HIMPORT_PREPARE && userPrepare !== undefined) { - // A rejected user PREPARE (e.g. duplicate field name) must not leave a registration - // that lazy prepare would replay forever. Version-guarded: a newer successful - // PREPARE must not be clobbered. Deleting through `discard()` also bumps - // discardCount, so sessions still holding an OLDER field list for this name - // reconcile it away instead of silently serving stale SETs. - const { name, version } = userPrepare; + // Roll back the optimistic registration. Version-guarded: a newer successful PREPARE + // that won a race must not be clobbered. + // • Rejected *replacement* (a prior field list existed) — the server keeps the old + // fieldset on reject, so restore it; dropping it would strand later SETs (on + // new/reconnected/pooled connections) with `no such fieldset` even though the last + // successful registration was still valid. + // • Rejected *fresh* PREPARE — remove via discard(), which also bumps discardCount so + // any session still holding a stale claim reconciles it away. + const { name, version, priorFields } = userPrepare; if (prepared.get(name) === version) prepared.delete(name); - if (registry.get(name)?.version === version) registry.discard(name); + if (registry.get(name)?.version === version) { + if (priorFields !== undefined) { + registry.set(name, priorFields); + } else { + registry.discard(name); + } + } } else if ( command === HIMPORT_SET && setName !== undefined && !retried && (err as Error)?.message?.includes?.('no such fieldset') && @@ -1480,12 +1512,22 @@ export default class RedisClient< prepared.delete(setName); return this.#executeHimport(client, command, parser, commandOptions, transformReply, true); } else if (userDiscard !== undefined) { - // A rejected user DISCARD/DISCARDALL leaves this session's server state unknown while - // the registry mutation stands (a discard is recorded user intent — other sessions - // reconcile off the count bump regardless). Restore this session's claims and knock - // the synced count back so the next HIMPORT command replays the discard here — the - // same recovery as a failed injected DISCARD. Absence-guarded: a re-PREPARE that won - // the race keeps its entry (its name is back in the registry, so reconcile skips it). + // A rejected user DISCARD/DISCARDALL, two cases: + // • Pre-enqueue failure (client closed/offline, aborted signal, full queue) — the + // command never reached the server, so the discard did not happen. Re-register the + // removed fieldsets (absence-guarded: a re-PREPARE that won the race keeps its newer + // entry) so shared duplicates/pools keep auto-preparing instead of failing with + // `no such fieldset` on a discard the server never saw. + // • Wire/server failure — this session's server state is unknown while the registry + // mutation stands (a discard is recorded user intent; other sessions reconcile off + // the count bump regardless). The registry stays removed and the synced-count knock + // below makes the next HIMPORT command replay the discard on this session. + // Either way, restore this session's optimistically-wiped claims. + if (isPreEnqueueError(err)) { + for (const [name, fields] of userDiscard.removed) { + if (registry.get(name) === undefined) registry.set(name, fields); + } + } for (const [name, version] of userDiscard.restore) { if (prepared.get(name) === undefined) prepared.set(name, version); } diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index 461a2c0319b..0d4d479a85e 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -126,7 +126,12 @@ export default class RedisClusterSlots< * 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 = new FieldsetRegistry(); + readonly #himportRegistry: FieldsetRegistry; + + /** The cluster-wide registry, exposed so `RedisCluster.duplicate()` can share it. */ + get himportRegistry() { + return this.#himportRegistry; + } smigratedSeqIdsSeen = new Set; #topologyRefreshPromise?: Promise; @@ -155,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); diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts index 43e82153847..973760d03cc 100644 --- a/packages/client/lib/cluster/index.ts +++ b/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>) { 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>; diff --git a/packages/client/lib/himport/registry.spec.ts b/packages/client/lib/himport/registry.spec.ts index 3fc659e2a02..4b696baa6e5 100644 --- a/packages/client/lib/himport/registry.spec.ts +++ b/packages/client/lib/himport/registry.spec.ts @@ -65,6 +65,23 @@ describe('FieldsetRegistry', () => { }); }); + describe('snapshot', () => { + it('returns every registration as name → field list', () => { + const registry = new FieldsetRegistry(); + registry.set('fs1', ['a', 'b']); + registry.set('fs2', ['c']); + + const snap = registry.snapshot(); + assert.deepEqual(snap.get('fs1'), ['a', 'b']); + assert.deepEqual(snap.get('fs2'), ['c']); + assert.equal(snap.size, 2); + }); + + it('is empty for an empty registry', () => { + assert.equal(new FieldsetRegistry().snapshot().size, 0); + }); + }); + describe('discard', () => { it('returns true and bumps discardCount for a registered name', () => { const registry = new FieldsetRegistry(); diff --git a/packages/client/lib/himport/registry.ts b/packages/client/lib/himport/registry.ts index 39fea4bc8c1..e6d37d87800 100644 --- a/packages/client/lib/himport/registry.ts +++ b/packages/client/lib/himport/registry.ts @@ -78,6 +78,16 @@ export class FieldsetRegistry { return this.#fieldsets.get(String(name)); } + /** + * Snapshot of every registration as name → field list. Used to roll back a `DISCARDALL` + * that failed before reaching the server, so shared duplicates/pools keep auto-preparing. + */ + snapshot(): Map> { + const out = new Map>(); + for (const [name, fs] of this.#fieldsets) out.set(name, fs.fields); + return out; + } + /** * Returns `true` iff the name was registered — this boolean IS the user-facing * `hImportDiscard` reply (registry-based, not the server session's). diff --git a/packages/client/lib/himport/transparency.spec.ts b/packages/client/lib/himport/transparency.spec.ts index 99be9f2727e..c6893ba3490 100644 --- a/packages/client/lib/himport/transparency.spec.ts +++ b/packages/client/lib/himport/transparency.spec.ts @@ -1,6 +1,7 @@ import { strict as assert } from 'node:assert'; import { once } from 'node:events'; import testUtils, { GLOBAL } from '../test-utils'; +import { AbortError } from '../errors'; /** * Integration tests for the HIMPORT transparency layer (`#executeHimport` + @@ -84,6 +85,43 @@ describe('HIMPORT transparency layer', () => { assert.equal(await client.hImportSet('key', 'fs', ['v1', 'v2']), 'OK'); }, GLOBAL.SERVERS.OPEN); + testUtils.testWithClient('rejected replacement PREPARE keeps the prior registration', async client => { + await client.hImportPrepare('fs', ['f1', 'f2']); + assert.equal(await client.hImportSet('key1', 'fs', ['v1', 'v2']), 'OK'); + + // Re-prepare an already-registered fieldset with an invalid (duplicate) field list. The + // server rejects it and keeps the existing fieldset, so the client must NOT drop the + // still-valid registration. + await assert.rejects( + client.hImportPrepare('fs', ['dup', 'dup']), + /duplicate field name/ + ); + + // Fresh session (empty server-side) so the SET can only succeed by lazy-preparing from a + // surviving registry entry — proving the prior registration was restored, not discarded. + await killClient(client); + assert.equal(await client.hImportSet('key2', 'fs', ['v1', 'v2']), 'OK'); + assert.deepEqual(await client.hGetAll('key2'), { f1: 'v1', f2: 'v2' }); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('pre-enqueue DISCARD failure keeps the registration', async client => { + await client.hImportPrepare('fs', ['f1']); + assert.equal(await client.hImportSet('key1', 'fs', ['v1']), 'OK'); + + // An already-aborted signal rejects the DISCARD before it reaches the wire — the server + // never saw it, so the discard must be fully rolled back (registry entry re-added). + const ac = new AbortController(); + ac.abort(); + await assert.rejects( + client.withAbortSignal(ac.signal).hImportDiscard('fs'), + AbortError + ); + + // The registration survives, so the fieldset still resolves. + assert.equal(await client.hImportSet('key2', 'fs', ['v1']), 'OK'); + assert.equal(await client.hGet('key2', 'f1'), 'v1'); + }, GLOBAL.SERVERS.OPEN); + testUtils.testWithClient('retries once when the session lost its state unobserved', async client => { await client.hImportPrepare('fs', ['f1']); await client.hImportSet('key1', 'fs', ['v1']); @@ -251,5 +289,19 @@ describe('HIMPORT transparency layer', () => { // sum, not a session count. assert.equal(await cluster.hImportDiscardAll(), 1); }, GLOBAL.CLUSTERS.OPEN); + + testUtils.testWithCluster('duplicate() shares HIMPORT registrations', async cluster => { + await cluster.hImportPrepare('fs', ['f1', 'f2']); + + // A duplicate made after PREPARE shares the parent's registry (matching the standalone + // duplicate-sharing guarantee), so a direct SET lazy-prepares from it per node. + const dup = await cluster.duplicate().connect(); + try { + assert.equal(await dup.hImportSet('key:{1}', 'fs', ['v1', 'v2']), 'OK'); + assert.deepEqual(await dup.hGetAll('key:{1}'), { f1: 'v1', f2: 'v2' }); + } finally { + await dup.close(); + } + }, GLOBAL.CLUSTERS.OPEN); }); }); From 994ac8d7acc6e65c829f28771991566ef06a828c Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Wed, 29 Jul 2026 17:24:35 +0300 Subject: [PATCH 3/3] fix(client): gate HIMPORT SET on its injected PREPARE and re-ASK on ASK chains Two more HIMPORT transparency-layer correctness fixes from PR review: - Tie the SET reply to its lazily-injected PREPARE. If that PREPARE is rejected while the connection stays alive (e.g. ACL denies the PREPARE subcommand), the SET would otherwise run against the stale server-side field list and silently store values under the wrong field names. Hold the PREPARE promise and, once the SET resolves, surface the PREPARE error and drop the session claim instead of returning the misleading OK. - Re-issue ASKING on ASK-redirect chains. A keyless PREPARE/DISCARD pipelined ahead of the SET consumes the one-shot ASKING flag, so the SET is redirected again until maxCommandRedirections. The cluster ASK handler now marks the command options (askRedirect), and the hook re-issues ASKING as the last prelude entry so it lands immediately before the SET. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/lib/client/commands-queue.ts | 7 +++ packages/client/lib/client/index.ts | 44 +++++++++++-- packages/client/lib/cluster/index.ts | 3 + .../client/lib/himport/transparency.spec.ts | 61 +++++++++++++++++++ 4 files changed, 110 insertions(+), 5 deletions(-) diff --git a/packages/client/lib/client/commands-queue.ts b/packages/client/lib/client/commands-queue.ts index c9a92b2f176..89c467c8d6e 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -24,6 +24,13 @@ export interface CommandOptions { * 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 { diff --git a/packages/client/lib/client/index.ts b/packages/client/lib/client/index.ts index 648c2a9fb01..84f124178d5 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -30,6 +30,7 @@ import HIMPORT_DISCARD from '../commands/HIMPORT_DISCARD'; import HIMPORT_DISCARDALL from '../commands/HIMPORT_DISCARDALL'; import HIMPORT_PREPARE from '../commands/HIMPORT_PREPARE'; import HIMPORT_SET from '../commands/HIMPORT_SET'; +import { ASKING_CMD } from '../commands/ASKING'; const noop = () => {}; @@ -1355,7 +1356,7 @@ export default class RedisClient< // Commands to reach the wire BEFORE the main command, in this order. Each carries the // rollback undoing its optimistic bookkeeping if the server rejects it. - const prelude: Array<{ args: Array, rollback: () => void }> = []; + const prelude: Array<{ args: Array, rollback: () => void, gateMain?: boolean }> = []; // -- Reconcile: replay discards this session may still be holding. A DISCARD must // precede a same-name SET on the wire, or the SET would write through the discarded @@ -1427,7 +1428,11 @@ export default class RedisClient< // permanently lose lazy re-prepare. rollback: () => { if (prepared.get(name) === entry.version) prepared.delete(name); - } + }, + // Gate the SET reply on this PREPARE: if it fails while the connection stays alive + // (e.g. ACL denies PREPARE), the SET would otherwise run against the stale + // server-side field list and silently write values under the wrong field names. + gateMain: true }); } } @@ -1460,21 +1465,35 @@ export default class RedisClient< prepared.clear(); } + // -- ASK chain: every keyless command in the prelude (a lazy PREPARE or a reconcile + // DISCARD) would consume the one-shot ASKING flag that the ASK handler set for this SET, + // leaving the SET to be redirected again until maxCommandRedirections. Re-issue ASKING as + // the final prelude entry so it lands immediately before the SET on the wire. + if (command === HIMPORT_SET && commandOptions?.askRedirect && prelude.length > 0) { + prelude.push({ args: [ASKING_CMD], rollback: () => {} }); + } + // -- Enqueue. Everything below runs in one synchronous tick, so the prelude and the // main command flush to the socket in a single write (the HLD-blessed pipelining). const send = () => client.sendCommand(parser.redisArgs, commandOptions); + // The version-bump PREPARE (if any) whose success the SET reply is gated on. + let mainGate: Promise | undefined; + const enqueue = (injection: { args: Array, rollback: () => void, gateMain?: boolean }) => { + const promise = client.sendCommand(injection.args, injectOpts); + promise.catch(injection.rollback); + if (injection.gateMain) mainGate = promise; + }; let mainPromise: Promise; if (effectiveAsap) { // asap unshifts, so consecutive front-insertions reverse: enqueue the main command // first, then the prelude back-to-front — the wire sees prelude order, then main. mainPromise = send(); for (let i = prelude.length - 1; i >= 0; i--) { - const injection = prelude[i]; - client.sendCommand(injection.args, injectOpts).catch(injection.rollback); + enqueue(prelude[i]); } } else { for (const injection of prelude) { - client.sendCommand(injection.args, injectOpts).catch(injection.rollback); + enqueue(injection); } mainPromise = send(); } @@ -1536,6 +1555,21 @@ export default class RedisClient< throw err; } + // The SET succeeded on the wire, but if its version-bump PREPARE was rejected while the + // connection stayed alive, the server applied the SET to a STALE field list (the older + // version this session still held) — a silent write under the wrong field names. Surface + // the PREPARE error instead of the misleading OK, and drop the now-unreliable session claim + // so the next SET re-prepares. (The `.catch(rollback)` already cleared the optimistic claim; + // this delete is idempotent.) + if (mainGate !== undefined) { + try { + await mainGate; + } catch (prepareErr) { + if (setName !== undefined) prepared.delete(setName); + throw prepareErr; + } + } + const finalReply = registryReply !== undefined ? registryReply : transformReply ? transformReply(reply, parser.preserve, commandOptions?.typeMapping) : reply; diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts index 973760d03cc..d9ebdedd4e5 100644 --- a/packages/client/lib/cluster/index.ts +++ b/packages/client/lib/cluster/index.ts @@ -490,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; diff --git a/packages/client/lib/himport/transparency.spec.ts b/packages/client/lib/himport/transparency.spec.ts index c6893ba3490..1edf41c2fc9 100644 --- a/packages/client/lib/himport/transparency.spec.ts +++ b/packages/client/lib/himport/transparency.spec.ts @@ -122,6 +122,67 @@ describe('HIMPORT transparency layer', () => { assert.equal(await client.hGet('key2', 'f1'), 'v1'); }, GLOBAL.SERVERS.OPEN); + testUtils.testWithClient('a failed injected PREPARE fails the SET instead of writing stale fields', async client => { + await client.hImportPrepare('fs', ['f1']); + // This connection now holds fs@v1 = [f1] server-side. + assert.equal(await client.hImportSet('key1', 'fs', ['x']), 'OK'); + + // A duplicate shares the registry. Re-preparing through it advances the registry version + // and prepares [g1] on the duplicate's own session — but THIS connection still holds the + // old [f1] version, so its next SET must lazily re-prepare. + const dup = await client.duplicate().connect(); + try { + await dup.hImportPrepare('fs', ['g1']); + + // Fail this connection's injected PREPARE while the socket stays alive. Without the gate + // the SET would run against the stale [f1] template and silently store 'y' under f1 + // instead of g1; the gate must surface the PREPARE error instead. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const self = (client as any)._self; + const original = self.sendCommand.bind(self); + self.sendCommand = (args: Array, opts?: unknown) => { + if (String(args[0]) === 'HIMPORT' && String(args[1]) === 'PREPARE') { + self.sendCommand = original; + return Promise.reject(new Error('simulated PREPARE failure')); + } + return original(args, opts); + }; + + await assert.rejects( + client.hImportSet('key2', 'fs', ['y']), + /simulated PREPARE failure/ + ); + } finally { + await dup.close(); + } + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('ASK chain re-issues ASKING after the injected PREPARE', async client => { + await client.hImportPrepare('fs', ['f1']); + // Fresh session so the next SET pipelines a lazy PREPARE ahead of itself. + await killClient(client); + + // Record the HIMPORT/ASKING wire order (handshake commands filtered out). + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const self = (client as any)._self; + const wire: Array = []; + const original = self.sendCommand.bind(self); + self.sendCommand = (args: Array, opts?: unknown) => { + const head = String(args[0]); + if (head === 'HIMPORT' || head === 'ASKING') { + wire.push([args[0], args[1]].filter(Boolean).map(String).join(' ')); + } + return original(args, opts); + }; + + // askRedirect mirrors what the cluster ASK handler sets. ASKING errors on a standalone + // server but is fire-and-forget, so it does not fail the SET. + await client.withCommandOptions({ askRedirect: true }).hImportSet('key', 'fs', ['v1']); + + // The re-issued ASKING lands immediately before the SET, keeping its one-shot flag. + assert.deepEqual(wire, ['HIMPORT PREPARE', 'ASKING', 'HIMPORT SET']); + }, GLOBAL.SERVERS.OPEN); + testUtils.testWithClient('retries once when the session lost its state unobserved', async client => { await client.hImportPrepare('fs', ['f1']); await client.hImportSet('key1', 'fs', ['v1']);