From e4c8b3fa57160fce57feb4875f451cce32f833c5 Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Tue, 28 Jul 2026 18:15:23 +0900 Subject: [PATCH 1/6] wip: batch command slot routing (#3366) --- packages/client/lib/client/commands-queue.ts | 12 ++ packages/client/lib/client/index.ts | 16 ++- .../client/lib/cluster/batch-routing.spec.ts | 60 ++++++++++ .../client/lib/cluster/cluster-slots.spec.ts | 63 +++++++++- packages/client/lib/cluster/cluster-slots.ts | 110 +++++++++++++++--- packages/client/lib/cluster/index.ts | 8 +- 6 files changed, 243 insertions(+), 26 deletions(-) create mode 100644 packages/client/lib/cluster/batch-routing.spec.ts diff --git a/packages/client/lib/client/commands-queue.ts b/packages/client/lib/client/commands-queue.ts index c9a92b2f176..9d19ae1daae 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -144,6 +144,18 @@ export default class RedisCommandsQueue { return this.#toWrite.length + this.#waitingForReply.length; } + /** + * The chainId of the most recently written command, if it belonged to a + * MULTI/pipeline chain. While this is set, any command still in `#toWrite` + * with a matching `chainId` is the tail of a chain that's already partially + * sent to the server - the rest of that chain is in `#waitingForReply` and + * out of view, so this tail must not be treated as a complete, relocatable + * chain (see `flushWaitingForReply`, which rejects it instead of resending it). + */ + get chainInExecution(): symbol | undefined { + return this.#chainInExecution; + } + constructor( respVersion: RespVersions, maxLength: number | null | undefined, diff --git a/packages/client/lib/client/index.ts b/packages/client/lib/client/index.ts index 9e491c9c396..efcf8a9b6b8 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -1508,7 +1508,8 @@ export default class RedisClient< */ async _executePipeline( commands: Array, - selectedDB?: number + selectedDB?: number, + slotNumber?: number ) { if (!this._self.#socket.isOpen) { return Promise.reject(new ClientClosedError()); @@ -1524,7 +1525,8 @@ export default class RedisClient< const traced = trace(CHANNELS.TRACE_COMMAND, () => this._self.#queue.addCommand(args, { chainId, - typeMapping: this._commandOptions?.typeMapping + typeMapping: this._commandOptions?.typeMapping, + slotNumber }), () => ({ ...this._self.#commandTraceContext(args), @@ -1564,7 +1566,8 @@ export default class RedisClient< */ async _executeMulti( commands: Array, - selectedDB?: number + selectedDB?: number, + slotNumber?: number ) { const dirtyWatch = this._self.#dirtyWatch; this._self.#dirtyWatch = undefined; @@ -1590,20 +1593,21 @@ export default class RedisClient< const typeMapping = this._commandOptions?.typeMapping; const chainId = Symbol('MULTI Chain'); const promises: Array> = [ - this._self.#queue.addCommand(['MULTI'], { chainId }), + this._self.#queue.addCommand(['MULTI'], { chainId, slotNumber }), ]; for (const { args } of commands) { promises.push( this._self.#queue.addCommand(args, { chainId, - typeMapping + typeMapping, + slotNumber }) ); } promises.push( - this._self.#queue.addCommand(['EXEC'], { chainId }) + this._self.#queue.addCommand(['EXEC'], { chainId, slotNumber }) ); this._self.#scheduleWrite(); diff --git a/packages/client/lib/cluster/batch-routing.spec.ts b/packages/client/lib/cluster/batch-routing.spec.ts new file mode 100644 index 00000000000..b76de4ffec5 --- /dev/null +++ b/packages/client/lib/cluster/batch-routing.spec.ts @@ -0,0 +1,60 @@ +import { strict as assert } from 'node:assert'; +import RedisCluster from '.'; + +describe('Cluster batch routing', () => { + it('passes the selected slot to MULTI execution', async () => { + const cluster = RedisCluster.create({ rootNodes: [] }); + let capturedSlotNumber: number | undefined; + + const fakeClient = { + _executeMulti: async (_commands: unknown, _selectedDB?: number, slotNumber?: number) => { + capturedSlotNumber = slotNumber; + return ['OK']; + } + }; + + cluster._slots.getClientAndSlotNumber = async () => ({ + client: fakeClient as never, + slotNumber: 123 + }); + + assert.deepEqual( + await cluster.multi() + .sendCommand(['SET', 'key', 'value'], { + firstKey: 'key', + isReadonly: false + }) + .exec(), + ['OK'] + ); + assert.equal(capturedSlotNumber, 123); + }); + + it('passes the selected slot to pipeline execution', async () => { + const cluster = RedisCluster.create({ rootNodes: [] }); + let capturedSlotNumber: number | undefined; + + const fakeClient = { + _executePipeline: async (_commands: unknown, _selectedDB?: number, slotNumber?: number) => { + capturedSlotNumber = slotNumber; + return ['OK']; + } + }; + + cluster._slots.getClientAndSlotNumber = async () => ({ + client: fakeClient as never, + slotNumber: 456 + }); + + assert.deepEqual( + await cluster.multi() + .sendCommand(['SET', 'key', 'value'], { + firstKey: 'key', + isReadonly: false + }) + .execAsPipeline(), + ['OK'] + ); + assert.equal(capturedSlotNumber, 456); + }); +}); diff --git a/packages/client/lib/cluster/cluster-slots.spec.ts b/packages/client/lib/cluster/cluster-slots.spec.ts index 9d03dfc723a..2b8ff075dd2 100644 --- a/packages/client/lib/cluster/cluster-slots.spec.ts +++ b/packages/client/lib/cluster/cluster-slots.spec.ts @@ -1,9 +1,30 @@ import { strict as assert } from 'node:assert'; import { EventEmitter } from 'node:events'; import { RedisClusterClientOptions } from './index'; -import RedisClusterSlots from './cluster-slots'; +import RedisClusterSlots, { groupCommandsByDestination } from './cluster-slots'; +import type { MasterNode, Shard } from './cluster-slots'; +import type { CommandToWrite } from '../client/commands-queue'; describe('RedisClusterSlots', () => { + function createCommand(slotNumber?: number) { + return { + args: ['CMD'], + slotNumber + } as CommandToWrite; + } + + function createMaster(address: string) { + return { + address + } as MasterNode< + Record, + Record, + Record, + 3, + Record + >; + } + describe('initialization', () => { describe('clientSideCache validation', () => { const mockEmit: EventEmitter['emit'] = () => true; @@ -53,4 +74,44 @@ describe('RedisClusterSlots', () => { slots.getRandomNode() }); }); + + describe('groupCommandsByDestination', () => { + it('groups commands by their slot owner instead of the fallback destination', () => { + const fallback = createMaster('fallback:6379'); + const slotOwner = createMaster('slot-owner:6379'); + const otherSlotOwner = createMaster('other-slot-owner:6379'); + const slots = [] as Array, + Record, + Record, + 3, + Record + >>; + slots[1] = { master: slotOwner }; + slots[2] = { master: otherSlotOwner }; + + const slotless = createCommand(); + const slotOne = createCommand(1); + const slotTwo = createCommand(2); + + const groups = groupCommandsByDestination( + [slotless, slotOne, slotTwo], + slots, + fallback + ); + + assert.deepEqual(groups.get(fallback), [slotless]); + assert.deepEqual(groups.get(slotOwner), [slotOne]); + assert.deepEqual(groups.get(otherSlotOwner), [slotTwo]); + }); + + it('falls back when a command has no known slot owner', () => { + const fallback = createMaster('fallback:6379'); + const command = createCommand(10); + + const groups = groupCommandsByDestination([command], [], fallback); + + assert.deepEqual(groups.get(fallback), [command]); + }); + }); }); diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index b40680531fb..d87311502b2 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -1,6 +1,7 @@ import type { RedisClusterClientOptions, RedisClusterOptions } from '.'; import { ClientClosedError, ClientOfflineError, DisconnectsClientError, RootNodesUnavailableError } from '../errors'; import RedisClient, { RedisClientOptions, RedisClientType } from '../client'; +import type { CommandToWrite } from '../client/commands-queue'; import { EventEmitter } from 'node:stream'; import { ChannelListeners, PUBSUB_TYPE, PubSubListeners, PubSubTypeListeners } from '../client/pub-sub'; import { RedisArgument, RedisFunctions, RedisModules, RedisScripts, RespVersions, TypeMapping, DEFAULT_RESP } from '../RESP/types'; @@ -94,6 +95,40 @@ type PubSubToResubscribe = Record< PubSubTypeListeners >; +export function groupCommandsByDestination< + M extends RedisModules, + F extends RedisFunctions, + S extends RedisScripts, + RESP extends RespVersions, + TYPE_MAPPING extends TypeMapping +>( + commands: CommandToWrite[], + slots: Array>, + fallback?: MasterNode +) { + const byDestination = new Map< + MasterNode, + CommandToWrite[] + >(); + + for (const command of commands) { + const destination = command.slotNumber === undefined ? + fallback : + slots[command.slotNumber]?.master ?? fallback; + + if (!destination) continue; + + const group = byDestination.get(destination); + if (group) { + group.push(command); + } else { + byDestination.set(destination, [command]); + } + } + + return byDestination; +} + export type OnShardedChannelMovedError = ( err: unknown, channel: string, @@ -385,12 +420,34 @@ export default class RedisClusterSlots< // 4. Extract commands for this destination's slots and prepend to destination queue const commandsForDestination = sourceNode.client._getQueue().extractCommandsForSlots(destinationSlots); - if (destMasterNode.client) { - destMasterNode.client._getQueue().prependCommandsToWrite(commandsForDestination); - dbgMaintenance(`[CSlots]: Extracted ${commandsForDestination.length} commands for ${destinationSlots.size} slots, prepended to ${destMasterNode.address}`); - } else { - sourceNode.client._getQueue().rejectCommands(commandsForDestination, new DisconnectsClientError()); - dbgMaintenance(`[CSlots]: Rejected ${commandsForDestination.length} commands for ${destinationSlots.size} slots because ${destMasterNode.address} has no client`); + if (commandsForDestination.length > 0) { + // Same in-flight chain safety as the full-node-loss path below: a + // chain already partially written to the socket can't be + // relocated as a fragment - its slotNumber matches this + // destination, but part of it is out of view in + // `#waitingForReply`. + const chainInExecution = sourceNode.client._getQueue().chainInExecution; + const inFlightChainTail = chainInExecution === undefined + ? [] + : commandsForDestination.filter(command => command.chainId === chainInExecution); + const relocatable = inFlightChainTail.length === 0 + ? commandsForDestination + : commandsForDestination.filter(command => command.chainId !== chainInExecution); + + if (inFlightChainTail.length > 0) { + sourceNode.client._getQueue().rejectCommands(inFlightChainTail, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${inFlightChainTail.length} commands from a chain already partially sent to the server, can't be safely relocated`); + } + + if (relocatable.length > 0) { + if (destMasterNode.client) { + destMasterNode.client._getQueue().prependCommandsToWrite(relocatable); + dbgMaintenance(`[CSlots]: Extracted ${relocatable.length} commands for ${destinationSlots.size} slots, prepended to ${destMasterNode.address}`); + } else { + sourceNode.client._getQueue().rejectCommands(relocatable, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${relocatable.length} commands for ${destinationSlots.size} slots because ${destMasterNode.address} has no client`); + } + } } // 5. Unpause destination @@ -430,17 +487,40 @@ export default class RedisClusterSlots< sourceNode.pubSub?.client._unpause(); } } else { - // Source has no slots left - move remaining slotless commands and cleanup + // Source has no slots left - move remaining commands to their current slot owners const remainingCommands = sourceNode.client._getQueue().extractAllCommands(); if (remainingCommands.length > 0) { - if (lastDestNode?.client) { - lastDestNode.client._getQueue().prependCommandsToWrite(remainingCommands); - // Trigger write scheduling since commands were added after destination was unpaused - lastDestNode.client._unpause(); - dbgMaintenance(`[CSlots]: Moved ${remainingCommands.length} remaining slotless commands to ${lastDestNode.address}`); - } else { - sourceNode.client._getQueue().rejectCommands(remainingCommands, new DisconnectsClientError()); - dbgMaintenance(`[CSlots]: Rejected ${remainingCommands.length} remaining commands because no destination client was available`); + // A chain (MULTI/pipeline) that's already partially written to the + // socket has the rest of its commands sitting in `#waitingForReply`, + // out of view. The tail still in `#toWrite` can't be safely + // relocated on its own - it would run as a fragment of the + // transaction on a different node than the part already sent. + const chainInExecution = sourceNode.client._getQueue().chainInExecution; + const inFlightChainTail = chainInExecution === undefined + ? [] + : remainingCommands.filter(command => command.chainId === chainInExecution); + const relocatable = inFlightChainTail.length === 0 + ? remainingCommands + : remainingCommands.filter(command => command.chainId !== chainInExecution); + + if (inFlightChainTail.length > 0) { + sourceNode.client._getQueue().rejectCommands(inFlightChainTail, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${inFlightChainTail.length} commands from a chain already partially sent to the server, can't be safely relocated`); + } + + if (relocatable.length > 0) { + const commandsByDestination = groupCommandsByDestination(relocatable, this.slots, lastDestNode); + for (const [destNode, commands] of commandsByDestination) { + if (destNode.client) { + destNode.client._getQueue().prependCommandsToWrite(commands); + // Trigger write scheduling since commands were added after destination was unpaused + destNode.client._unpause(); + dbgMaintenance(`[CSlots]: Moved ${commands.length} remaining commands to ${destNode.address}`); + } else { + sourceNode.client._getQueue().rejectCommands(commands, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${commands.length} remaining commands because ${destNode.address} has no client`); + } + } } } diff --git a/packages/client/lib/cluster/index.ts b/packages/client/lib/cluster/index.ts index 43e82153847..eb0977449b3 100644 --- a/packages/client/lib/cluster/index.ts +++ b/packages/client/lib/cluster/index.ts @@ -596,12 +596,12 @@ export default class RedisCluster< type Multi = new (...args: ConstructorParameters) => RedisClusterMultiCommandType<[], M, F, S, RESP, TYPE_MAPPING>; return new (this as this & { Multi: Multi }).Multi( async (firstKey, isReadonly, commands) => { - const { client } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly); - return client._executeMulti(commands); + const { client, slotNumber } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly); + return client._executeMulti(commands, undefined, slotNumber); }, async (firstKey, isReadonly, commands) => { - const { client } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly); - return client._executePipeline(commands); + const { client, slotNumber } = await this._self._slots.getClientAndSlotNumber(firstKey, isReadonly); + return client._executePipeline(commands, undefined, slotNumber); }, routing, this._commandOptions?.typeMapping, From 2a08d72bf916a1d3d03dc7dee920a978de4baf3d Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Tue, 28 Jul 2026 18:34:11 +0900 Subject: [PATCH 2/6] fix(cluster): don't silently drop unrouted commands, scope in-flight chain guard to full-node-loss only - groupCommandsByDestination() now returns {byDestination, unrouted} instead of silently continue-ing past commands with neither a slot owner nor a fallback destination. The full-node-loss caller now rejects unrouted commands explicitly instead of leaving them extracted-but-never-settled (the same failure mode #3365 fixed). - Removed the in-flight-chain-tail rejection from the partial-slot- migration path. That source connection isn't destroyed - it keeps serving its remaining slots - so rejecting the queued tail doesn't clear whatever transaction state the server still thinks is open on it. The full-node-loss path keeps the guard, since that connection is destroyed right after. - Renamed a chainInExecution test whose name didn't match its assertion. --- .../client/lib/client/commands-queue.spec.ts | 47 +++++++++++++++ .../client/lib/cluster/cluster-slots.spec.ts | 28 +++++++-- packages/client/lib/cluster/cluster-slots.ts | 60 ++++++++++--------- 3 files changed, 100 insertions(+), 35 deletions(-) diff --git a/packages/client/lib/client/commands-queue.spec.ts b/packages/client/lib/client/commands-queue.spec.ts index 533b3de0fd7..73c577e2fa1 100644 --- a/packages/client/lib/client/commands-queue.spec.ts +++ b/packages/client/lib/client/commands-queue.spec.ts @@ -29,6 +29,53 @@ describe('RedisCommandsQueue', () => { }); }); + describe('chainInExecution', () => { + it('is undefined before anything is written, and matches a chain once part of it is sent', () => { + const queue = createQueue(); + const chainId = Symbol('MULTI Chain'); + + assert.strictEqual(queue.chainInExecution, undefined); + + queue.addCommand(['MULTI'], { chainId }).catch(() => {}); + queue.addCommand(['SET', 'k', 'v'], { chainId }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId }).catch(() => {}); + + // Nothing sent yet - the whole chain is still just sitting in #toWrite. + assert.strictEqual(queue.chainInExecution, undefined); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" MULTI + writer.next(); // "sends" SET + + // Two of the three commands are now in #waitingForReply (out of view); + // chainInExecution should point at this chain, and the one command + // still in #toWrite (EXEC) should carry a matching chainId - that's + // the tail-of-an-in-flight-chain signal cluster-slots.ts relies on. + assert.strictEqual(queue.chainInExecution, chainId); + + const remaining = queue.extractAllCommands(); + assert.strictEqual(remaining.length, 1); + assert.strictEqual(remaining[0].chainId, queue.chainInExecution); + }); + + it('leaves a finished chain with no queued tail to relocate', () => { + const queue = createQueue(); + const chainId = Symbol('MULTI Chain'); + + queue.addCommand(['MULTI'], { chainId }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId }).catch(() => {}); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" MULTI + writer.next(); // "sends" EXEC - whole chain is now in #waitingForReply + + // The whole chain was sent, so nothing of it remains in #toWrite - + // extractAllCommands has nothing left to (mis)classify as its tail. + assert.strictEqual(queue.chainInExecution, chainId); + assert.strictEqual(queue.extractAllCommands().length, 0); + }); + }); + describe('addCommand', () => { it('does not keep a command if timeout listener setup fails', async () => { const queue = createQueue(); diff --git a/packages/client/lib/cluster/cluster-slots.spec.ts b/packages/client/lib/cluster/cluster-slots.spec.ts index 2b8ff075dd2..cd0cc701bf9 100644 --- a/packages/client/lib/cluster/cluster-slots.spec.ts +++ b/packages/client/lib/cluster/cluster-slots.spec.ts @@ -94,24 +94,40 @@ describe('RedisClusterSlots', () => { const slotOne = createCommand(1); const slotTwo = createCommand(2); - const groups = groupCommandsByDestination( + const { byDestination, unrouted } = groupCommandsByDestination( [slotless, slotOne, slotTwo], slots, fallback ); - assert.deepEqual(groups.get(fallback), [slotless]); - assert.deepEqual(groups.get(slotOwner), [slotOne]); - assert.deepEqual(groups.get(otherSlotOwner), [slotTwo]); + assert.deepEqual(byDestination.get(fallback), [slotless]); + assert.deepEqual(byDestination.get(slotOwner), [slotOne]); + assert.deepEqual(byDestination.get(otherSlotOwner), [slotTwo]); + assert.deepEqual(unrouted, []); }); it('falls back when a command has no known slot owner', () => { const fallback = createMaster('fallback:6379'); const command = createCommand(10); - const groups = groupCommandsByDestination([command], [], fallback); + const { byDestination, unrouted } = groupCommandsByDestination([command], [], fallback); - assert.deepEqual(groups.get(fallback), [command]); + assert.deepEqual(byDestination.get(fallback), [command]); + assert.deepEqual(unrouted, []); + }); + + it('reports commands as unrouted instead of dropping them when there is no fallback either', () => { + const slotless = createCommand(); + const unknownSlot = createCommand(10); + + const { byDestination, unrouted } = groupCommandsByDestination( + [slotless, unknownSlot], + [], + undefined + ); + + assert.strictEqual(byDestination.size, 0); + assert.deepEqual(unrouted, [slotless, unknownSlot]); }); }); }); diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index d87311502b2..6e8eaaa1136 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -110,13 +110,21 @@ export function groupCommandsByDestination< MasterNode, CommandToWrite[] >(); + // Commands with neither a resolvable slot owner nor a fallback - the + // caller must settle these explicitly (e.g. reject them) instead of + // silently dropping them, since they were already removed from the + // source queue by the time this is called. + const unrouted: CommandToWrite[] = []; for (const command of commands) { const destination = command.slotNumber === undefined ? fallback : slots[command.slotNumber]?.master ?? fallback; - if (!destination) continue; + if (!destination) { + unrouted.push(command); + continue; + } const group = byDestination.get(destination); if (group) { @@ -126,7 +134,7 @@ export function groupCommandsByDestination< } } - return byDestination; + return { byDestination, unrouted }; } export type OnShardedChannelMovedError = ( @@ -419,34 +427,22 @@ export default class RedisClusterSlots< dbgMaintenance(`[CSlots]: Updated slots to point to destination ${destMasterNode.address}. Sample slots: ${Array.from(slots).slice(0, 5).join(', ')}${slots.length > 5 ? '...' : ''}`); // 4. Extract commands for this destination's slots and prepend to destination queue + // + // Note: unlike the full-node-loss path below, this doesn't special-case + // a chain already partially written to the socket. The source + // connection here isn't destroyed - it keeps serving its remaining + // slots - so rejecting just the queued tail wouldn't undo whatever the + // server thinks is still an open transaction on that connection. + // Fixing that would mean resetting the source connection, which is a + // bigger change than this fix's scope. const commandsForDestination = sourceNode.client._getQueue().extractCommandsForSlots(destinationSlots); if (commandsForDestination.length > 0) { - // Same in-flight chain safety as the full-node-loss path below: a - // chain already partially written to the socket can't be - // relocated as a fragment - its slotNumber matches this - // destination, but part of it is out of view in - // `#waitingForReply`. - const chainInExecution = sourceNode.client._getQueue().chainInExecution; - const inFlightChainTail = chainInExecution === undefined - ? [] - : commandsForDestination.filter(command => command.chainId === chainInExecution); - const relocatable = inFlightChainTail.length === 0 - ? commandsForDestination - : commandsForDestination.filter(command => command.chainId !== chainInExecution); - - if (inFlightChainTail.length > 0) { - sourceNode.client._getQueue().rejectCommands(inFlightChainTail, new DisconnectsClientError()); - dbgMaintenance(`[CSlots]: Rejected ${inFlightChainTail.length} commands from a chain already partially sent to the server, can't be safely relocated`); - } - - if (relocatable.length > 0) { - if (destMasterNode.client) { - destMasterNode.client._getQueue().prependCommandsToWrite(relocatable); - dbgMaintenance(`[CSlots]: Extracted ${relocatable.length} commands for ${destinationSlots.size} slots, prepended to ${destMasterNode.address}`); - } else { - sourceNode.client._getQueue().rejectCommands(relocatable, new DisconnectsClientError()); - dbgMaintenance(`[CSlots]: Rejected ${relocatable.length} commands for ${destinationSlots.size} slots because ${destMasterNode.address} has no client`); - } + if (destMasterNode.client) { + destMasterNode.client._getQueue().prependCommandsToWrite(commandsForDestination); + dbgMaintenance(`[CSlots]: Extracted ${commandsForDestination.length} commands for ${destinationSlots.size} slots, prepended to ${destMasterNode.address}`); + } else { + sourceNode.client._getQueue().rejectCommands(commandsForDestination, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${commandsForDestination.length} commands for ${destinationSlots.size} slots because ${destMasterNode.address} has no client`); } } @@ -509,7 +505,13 @@ export default class RedisClusterSlots< } if (relocatable.length > 0) { - const commandsByDestination = groupCommandsByDestination(relocatable, this.slots, lastDestNode); + const { byDestination: commandsByDestination, unrouted } = groupCommandsByDestination(relocatable, this.slots, lastDestNode); + + if (unrouted.length > 0) { + sourceNode.client._getQueue().rejectCommands(unrouted, new DisconnectsClientError()); + dbgMaintenance(`[CSlots]: Rejected ${unrouted.length} commands with no resolvable slot owner and no fallback destination`); + } + for (const [destNode, commands] of commandsByDestination) { if (destNode.client) { destNode.client._getQueue().prependCommandsToWrite(commands); From c0c7e52313a4e7cdbab68ef4178ffa12d57f31b3 Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Tue, 28 Jul 2026 18:45:18 +0900 Subject: [PATCH 3/6] fix(cluster): don't extract an in-flight chain's queued tail in extractCommandsForSlots Once MULTI/pipeline commands carry a slotNumber, extractCommandsForSlots() (used during partial slot migration) started matching them - including a chain whose head was already written to the socket, with only its tail still queued. Relocating just that tail to another node would split the transaction across two connections, and unlike full node loss, this source connection isn't going away. Skip a command when its chainId matches chainInExecution, guarded so an ordinary non-chain command (chainId undefined) isn't mistaken for the tail of nothing (chainInExecution also undefined). The tail is left queued and sent to the original node as before; if the key has since moved, that's a normal MOVED/ASK error on the existing retry path. Caught this by reproducing the exact sequence directly against the queue: send MULTI, then call extractCommandsForSlots() and confirm the queued SET/EXEC come back. --- .../client/lib/client/commands-queue.spec.ts | 27 +++++++++++++++++++ packages/client/lib/client/commands-queue.ts | 12 ++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/packages/client/lib/client/commands-queue.spec.ts b/packages/client/lib/client/commands-queue.spec.ts index 73c577e2fa1..0d6e8dfe88a 100644 --- a/packages/client/lib/client/commands-queue.spec.ts +++ b/packages/client/lib/client/commands-queue.spec.ts @@ -76,6 +76,33 @@ describe('RedisCommandsQueue', () => { }); }); + describe('extractCommandsForSlots', () => { + it('leaves the queued tail of an in-flight chain in place instead of extracting it', () => { + const queue = createQueue(); + const chainId = Symbol('MULTI Chain'); + + queue.addCommand(['MULTI'], { chainId, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['SET', 'k', 'v'], { chainId, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId, slotNumber: 1 }).catch(() => {}); + // An unrelated command on the same slot, queued after the chain, should + // still be extracted normally - only the in-flight chain's own tail is + // held back. + queue.addCommand(['GET', 'k'], { slotNumber: 1 }).catch(() => {}); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" MULTI - SET and EXEC are still queued behind it + + const extracted = queue.extractCommandsForSlots(new Set([1])); + + assert.deepStrictEqual( + extracted.map(command => command.args?.[0]), + ['GET'], + ); + // SET and EXEC are still sitting in #toWrite, untouched. + assert.strictEqual(queue.pendingCount, 3); + }); + }); + describe('addCommand', () => { it('does not keep a command if timeout listener setup fails', async () => { const queue = createQueue(); diff --git a/packages/client/lib/client/commands-queue.ts b/packages/client/lib/client/commands-queue.ts index 9d19ae1daae..5c164fd21dc 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -703,6 +703,15 @@ export default class RedisCommandsQueue { * Extracts commands for the given slots from the toWrite queue. * Some commands don't have "slotNumber", which means they are not designated to particular slot/node. * We ignore those. + * + * A command whose chainId matches `chainInExecution` is the queued tail of a + * chain (MULTI/pipeline) whose head was already written to this socket - the + * rest is in `#waitingForReply`, out of view. It's left in place rather than + * extracted: relocating just the tail would split the transaction across two + * connections, and this connection isn't going away (unlike a full node + * loss), so it'll be sent here as originally queued. If the key has by then + * moved to another node, that surfaces as a normal MOVED/ASK error on the + * existing retry path, not something this method needs to prevent. */ extractCommandsForSlots(slots: Set): CommandToWrite[] { const result: CommandToWrite[] = []; @@ -710,7 +719,8 @@ export default class RedisCommandsQueue { while (current !== undefined) { if ( current.value.slotNumber !== undefined && - slots.has(current.value.slotNumber) + slots.has(current.value.slotNumber) && + (this.#chainInExecution === undefined || current.value.chainId !== this.#chainInExecution) ) { const command = current.value; if (command.abort) { From 891a00d94540dbd5595394e4fd60e0b151400128 Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Wed, 29 Jul 2026 10:10:30 +0900 Subject: [PATCH 4/6] fix(cluster): stop extraction at an in-flight chain's tail instead of skipping past it extractCommandsForSlots() was only skipping the in-flight chain's own tail commands, then continuing to extract anything queued after them that matched the requested slots. Since commands on one connection are sent in strict queue order, relocating a later command to a different connection while the chain ahead of it is still pending could let it run out of order - e.g. a GET queued right after an in-flight MULTI/SET/EXEC could execute on the new node before the transaction completes on the old one, reading a stale value. Once the scan reaches the in-flight tail, stop extracting entirely - everything from that point stays on this connection, in original order. Commands queued ahead of the chain aren't affected: by the time any part of a chain is in flight, anything queued before it has necessarily already been sent (this queue processes strictly FIFO), so there's nothing earlier left in #toWrite to reorder. Also corrected a comment in cluster-slots.ts that said the partial- migration path doesn't special-case in-flight chains, when the special- casing actually lives inside extractCommandsForSlots() itself. --- .../client/lib/client/commands-queue.spec.ts | 37 +++++++++++--- packages/client/lib/client/commands-queue.ts | 50 ++++++++++--------- packages/client/lib/cluster/cluster-slots.ts | 14 +++--- 3 files changed, 63 insertions(+), 38 deletions(-) diff --git a/packages/client/lib/client/commands-queue.spec.ts b/packages/client/lib/client/commands-queue.spec.ts index 0d6e8dfe88a..4b621be06e4 100644 --- a/packages/client/lib/client/commands-queue.spec.ts +++ b/packages/client/lib/client/commands-queue.spec.ts @@ -77,29 +77,52 @@ describe('RedisCommandsQueue', () => { }); describe('extractCommandsForSlots', () => { - it('leaves the queued tail of an in-flight chain in place instead of extracting it', () => { + it('leaves the queued tail of an in-flight chain and everything after it in place', () => { const queue = createQueue(); const chainId = Symbol('MULTI Chain'); queue.addCommand(['MULTI'], { chainId, slotNumber: 1 }).catch(() => {}); queue.addCommand(['SET', 'k', 'v'], { chainId, slotNumber: 1 }).catch(() => {}); queue.addCommand(['EXEC'], { chainId, slotNumber: 1 }).catch(() => {}); - // An unrelated command on the same slot, queued after the chain, should - // still be extracted normally - only the in-flight chain's own tail is - // held back. + // Queued after the chain, on the same connection. If this were + // relocated to another node while the transaction is still pending + // here, it could run before the transaction completes - reading 'k' + // before SET applies, even though it was queued after EXEC. queue.addCommand(['GET', 'k'], { slotNumber: 1 }).catch(() => {}); const writer = queue.commandsToWrite(); - writer.next(); // "sends" MULTI - SET and EXEC are still queued behind it + writer.next(); // "sends" MULTI - SET, EXEC and GET are still queued behind it const extracted = queue.extractCommandsForSlots(new Set([1])); + // Nothing is extracted: reaching the in-flight chain's tail stops the + // scan entirely, so GET stays behind it in queue order too. + assert.deepStrictEqual(extracted, []); + assert.strictEqual(queue.pendingCount, 4); + }); + + it('is unaffected by an already-fully-sent chain', () => { + const queue = createQueue(); + const chainId = Symbol('MULTI Chain'); + + queue.addCommand(['MULTI'], { chainId, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['GET', 'k'], { slotNumber: 1 }).catch(() => {}); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" MULTI + writer.next(); // "sends" EXEC - the whole chain is now in #waitingForReply + + const extracted = queue.extractCommandsForSlots(new Set([1])); + + // chainInExecution still points at this chain (nothing has been sent + // since), but none of its commands remain in #toWrite to (mis)match - + // the trailing GET is extracted normally. + assert.strictEqual(queue.chainInExecution, chainId); assert.deepStrictEqual( extracted.map(command => command.args?.[0]), ['GET'], ); - // SET and EXEC are still sitting in #toWrite, untouched. - assert.strictEqual(queue.pendingCount, 3); }); }); diff --git a/packages/client/lib/client/commands-queue.ts b/packages/client/lib/client/commands-queue.ts index 5c164fd21dc..2d612fe6ed5 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -706,37 +706,39 @@ export default class RedisCommandsQueue { * * A command whose chainId matches `chainInExecution` is the queued tail of a * chain (MULTI/pipeline) whose head was already written to this socket - the - * rest is in `#waitingForReply`, out of view. It's left in place rather than - * extracted: relocating just the tail would split the transaction across two - * connections, and this connection isn't going away (unlike a full node - * loss), so it'll be sent here as originally queued. If the key has by then - * moved to another node, that surfaces as a normal MOVED/ASK error on the - * existing retry path, not something this method needs to prevent. + * rest is in `#waitingForReply`, out of view. Relocating just the tail would + * split the transaction across two connections, and this connection isn't + * going away (unlike a full node loss), so extraction stops there entirely: + * that command and everything still queued behind it stay on this + * connection, preserving the order they were originally sent in, and get + * sent here as originally queued. If a key has by then moved to another + * node, that surfaces as a normal MOVED/ASK error on the existing retry + * path, not something this method needs to prevent. */ extractCommandsForSlots(slots: Set): CommandToWrite[] { const result: CommandToWrite[] = []; let current = this.#toWrite.head; while (current !== undefined) { - if ( - current.value.slotNumber !== undefined && - slots.has(current.value.slotNumber) && - (this.#chainInExecution === undefined || current.value.chainId !== this.#chainInExecution) - ) { - const command = current.value; - if (command.abort) { - RedisCommandsQueue.#removeAbortListener(command); - } - if (command.timeout) { - RedisCommandsQueue.#removeTimeoutListener(command); - } - result.push(command); - const toRemove = current; - current = current.next; - this.#toWrite.remove(toRemove); - } else { - // Move to next node even if we don't extract this command + if (current.value.slotNumber === undefined || !slots.has(current.value.slotNumber)) { current = current.next; + continue; + } + + if (this.#chainInExecution !== undefined && current.value.chainId === this.#chainInExecution) { + break; + } + + const command = current.value; + if (command.abort) { + RedisCommandsQueue.#removeAbortListener(command); + } + if (command.timeout) { + RedisCommandsQueue.#removeTimeoutListener(command); } + result.push(command); + const toRemove = current; + current = current.next; + this.#toWrite.remove(toRemove); } return result; } diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index 6e8eaaa1136..f6a4789a1a3 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -428,13 +428,13 @@ export default class RedisClusterSlots< // 4. Extract commands for this destination's slots and prepend to destination queue // - // Note: unlike the full-node-loss path below, this doesn't special-case - // a chain already partially written to the socket. The source - // connection here isn't destroyed - it keeps serving its remaining - // slots - so rejecting just the queued tail wouldn't undo whatever the - // server thinks is still an open transaction on that connection. - // Fixing that would mean resetting the source connection, which is a - // bigger change than this fix's scope. + // Note: unlike the full-node-loss path below, this doesn't reject an + // in-flight chain's queued tail - the source connection here isn't + // destroyed, it keeps serving its remaining slots, so rejecting the + // tail wouldn't undo whatever the server thinks is still an open + // transaction on that connection. Instead, extractCommandsForSlots() + // itself leaves an in-flight chain's tail (and anything queued after + // it) on source to preserve ordering; see its docstring. const commandsForDestination = sourceNode.client._getQueue().extractCommandsForSlots(destinationSlots); if (commandsForDestination.length > 0) { if (destMasterNode.client) { From cc8817a118de8f6dce4988fb7e392cdbe5676d51 Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Wed, 29 Jul 2026 10:22:09 +0900 Subject: [PATCH 5/6] docs(cluster): correct extractCommandsForSlots comment about MOVED/ASK handling MULTI/pipeline execution doesn't go through cluster's redirect-and-retry loop (that's sendCommand-only, via _execute()) - calling it an 'existing retry path' overstated what actually happens for batches. It's a normal error surfaced to the caller, not something automatically retried. --- packages/client/lib/client/commands-queue.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/client/lib/client/commands-queue.ts b/packages/client/lib/client/commands-queue.ts index 2d612fe6ed5..fc84e6447ae 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -712,8 +712,11 @@ export default class RedisCommandsQueue { * that command and everything still queued behind it stay on this * connection, preserving the order they were originally sent in, and get * sent here as originally queued. If a key has by then moved to another - * node, that surfaces as a normal MOVED/ASK error on the existing retry - * path, not something this method needs to prevent. + * node, that surfaces as a normal MOVED/ASK error through this method's + * caller - unlike single commands, MULTI/pipeline execution doesn't go + * through cluster's redirect-and-retry loop, so this isn't automatically + * retried, but that's an existing, separate limitation this method isn't + * responsible for preventing. */ extractCommandsForSlots(slots: Set): CommandToWrite[] { const result: CommandToWrite[] = []; From 2b3e959b9ae1cf003ee7147ead5cb747610b9305 Mon Sep 17 00:00:00 2001 From: GiHoon1123 Date: Thu, 30 Jul 2026 10:47:38 +0900 Subject: [PATCH 6/6] test(cluster): cover in-flight chain isolation and full-node-loss head rejection Addresses review feedback on PR #3377 (nkaradzhov): - extractCommandsForSlots now has a test proving a second, fully-queued chain behind an in-flight one on a different slot still relocates atomically and isn't mistaken for the in-flight chain's tail. - A full-node-loss test proves the already-sent head of an in-flight chain (in #waitingForReply) rejects alongside its queued tail, so the transaction fails consistently instead of half-committing. - Extracted the full-node-loss path's in-flight-tail/relocatable split into splitInFlightChainTail() (mirroring the existing groupCommandsByDestination pattern) so it can be unit tested directly, and added a test proving it separates an in-flight chain's tail from an unrelated, fully-queued chain queued right behind it. --- .../client/lib/client/commands-queue.spec.ts | 86 +++++++++++++++++++ .../client/lib/cluster/cluster-slots.spec.ts | 48 ++++++++++- packages/client/lib/cluster/cluster-slots.ts | 33 +++++-- 3 files changed, 160 insertions(+), 7 deletions(-) diff --git a/packages/client/lib/client/commands-queue.spec.ts b/packages/client/lib/client/commands-queue.spec.ts index 4b621be06e4..50688ac1115 100644 --- a/packages/client/lib/client/commands-queue.spec.ts +++ b/packages/client/lib/client/commands-queue.spec.ts @@ -124,6 +124,49 @@ describe('RedisCommandsQueue', () => { ['GET'], ); }); + + it('extracts a second, fully-queued chain atomically without mistaking it for the in-flight chain\'s tail', () => { + const queue = createQueue(); + const chainA = Symbol('Chain A (in-flight, different slot)'); + const chainB = Symbol('Chain B (fully queued, migrating slot)'); + + // Chain A is in flight on a slot that isn't migrating - its queued + // tail (SET, EXEC) carries a chainId that will equal chainInExecution. + queue.addCommand(['MULTI'], { chainId: chainA, slotNumber: 5 }).catch(() => {}); + queue.addCommand(['SET', 'a', '1'], { chainId: chainA, slotNumber: 5 }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId: chainA, slotNumber: 5 }).catch(() => {}); + + // Chain B is queued entirely behind chain A, on the slot that IS + // migrating. None of its commands have been sent, so none of them + // carry chainInExecution's id - it's a distinct chain, not a + // continuation of chain A's tail. + queue.addCommand(['MULTI'], { chainId: chainB, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['SET', 'b', '2'], { chainId: chainB, slotNumber: 1 }).catch(() => {}); + queue.addCommand(['EXEC'], { chainId: chainB, slotNumber: 1 }).catch(() => {}); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" chain A's MULTI - chainInExecution now points at chain A + + const extracted = queue.extractCommandsForSlots(new Set([1])); + + // Chain A's tail lives on slot 5, so it's skipped over (not in the + // migrating slot set) without ever triggering the in-flight-tail + // guard. Chain B, entirely on slot 1, is then extracted as a whole - + // its distinct chainId never matches chainInExecution, so it's never + // mistaken for chain A's tail and relocates atomically, in order. + assert.deepStrictEqual( + extracted.map(command => command.args?.[0]), + ['MULTI', 'SET', 'EXEC'], + ); + assert.ok(extracted.every(command => command.chainId === chainB)); + + // Chain A's tail stays behind, untouched, on its own slot. + const remaining = queue.extractAllCommands(); + assert.deepStrictEqual( + remaining.map(command => command.args?.[0]), + ['SET', 'EXEC'], + ); + }); }); describe('addCommand', () => { @@ -169,6 +212,49 @@ describe('RedisCommandsQueue', () => { }); }); + describe('full node loss (in-flight chain)', () => { + // Mirrors cluster-slots.ts's full-node-loss path: extractAllCommands() + // pulls the in-flight chain's queued tail out of #toWrite and rejects it + // explicitly, then the source client is destroyed, which calls + // flushAll() and rejects whatever's left in #waitingForReply - the + // chain's already-sent head. Both halves need to reject; otherwise the + // transaction half-commits: the server already applied the head against + // a connection this client is abandoning, while the tail is rejected as + // never sent. + it('rejects the already-sent head of an in-flight chain along with its queued tail', async () => { + const queue = createQueue(); + const chainId = Symbol('MULTI Chain'); + + const multiPromise = queue.addCommand(['MULTI'], { chainId }); + const setPromise = queue.addCommand(['SET', 'k', 'v'], { chainId }); + const execPromise = queue.addCommand(['EXEC'], { chainId }); + [multiPromise, setPromise, execPromise].forEach(promise => promise.catch(() => {})); + + const writer = queue.commandsToWrite(); + writer.next(); // "sends" MULTI - now in #waitingForReply + writer.next(); // "sends" SET - now in #waitingForReply + + // EXEC is still the queued tail in #toWrite. + const remaining = queue.extractAllCommands(); + assert.strictEqual(remaining.length, 1); + assert.strictEqual(remaining[0].chainId, queue.chainInExecution); + + // cluster-slots.ts rejects the queued tail explicitly, since it can't + // be safely relocated to another node... + queue.rejectCommands(remaining, new DisconnectsClientError()); + + // ...then destroy()'s call to flushAll() rejects whatever's left in + // #waitingForReply - the already-sent MULTI and SET. + queue.flushAll(new DisconnectsClientError()); + + await Promise.all([ + assert.rejects(multiPromise, DisconnectsClientError), + assert.rejects(setPromise, DisconnectsClientError), + assert.rejects(execPromise, DisconnectsClientError), + ]); + }); + }); + describe('prependCommandsToWrite', () => { it('rebinds an abort listener so it removes the command from its new queue', async () => { diff --git a/packages/client/lib/cluster/cluster-slots.spec.ts b/packages/client/lib/cluster/cluster-slots.spec.ts index cd0cc701bf9..e892607c392 100644 --- a/packages/client/lib/cluster/cluster-slots.spec.ts +++ b/packages/client/lib/cluster/cluster-slots.spec.ts @@ -1,7 +1,7 @@ import { strict as assert } from 'node:assert'; import { EventEmitter } from 'node:events'; import { RedisClusterClientOptions } from './index'; -import RedisClusterSlots, { groupCommandsByDestination } from './cluster-slots'; +import RedisClusterSlots, { groupCommandsByDestination, splitInFlightChainTail } from './cluster-slots'; import type { MasterNode, Shard } from './cluster-slots'; import type { CommandToWrite } from '../client/commands-queue'; @@ -130,4 +130,50 @@ describe('RedisClusterSlots', () => { assert.deepEqual(unrouted, [slotless, unknownSlot]); }); }); + + describe('splitInFlightChainTail', () => { + function createChainCommand(chainId: symbol, slotNumber?: number) { + return { args: ['CMD'], slotNumber, chainId } as CommandToWrite; + } + + // Mirrors the full-node-loss path in cluster-slots.ts: extractAllCommands() + // pulls everything out of a dying node's queue regardless of slot, then + // this function has to tell apart the in-flight chain's own queued tail + // (whose head is already sent, out of view in #waitingForReply - can't be + // safely relocated) from an unrelated, fully-queued chain that happens to + // be sitting right behind it and is safe to relocate whole. + it('separates only the in-flight chain\'s tail and leaves an unrelated, fully-queued chain relocatable', () => { + const chainA = Symbol('Chain A (in-flight)'); + const chainB = Symbol('Chain B (fully queued, never sent)'); + + const chainATail = [createChainCommand(chainA, 5), createChainCommand(chainA, 5)]; + const chainBCommands = [ + createChainCommand(chainB, 1), + createChainCommand(chainB, 1), + createChainCommand(chainB, 1), + ]; + + const { inFlightChainTail, relocatable } = splitInFlightChainTail( + [...chainATail, ...chainBCommands], + chainA, + ); + + // The in-flight chain's tail is set apart, not relocatable - the + // caller rejects it... + assert.deepEqual(inFlightChainTail, chainATail); + // ...and chain B, despite queuing right behind it, isn't mistaken for + // part of it - it comes back whole, ready to relocate atomically. + assert.deepEqual(relocatable, chainBCommands); + }); + + it('treats every command as relocatable when nothing is in flight', () => { + const chainB = Symbol('Chain B'); + const commands = [createChainCommand(chainB, 1), createChainCommand(chainB, 1)]; + + const { inFlightChainTail, relocatable } = splitInFlightChainTail(commands, undefined); + + assert.deepEqual(inFlightChainTail, []); + assert.deepEqual(relocatable, commands); + }); + }); }); diff --git a/packages/client/lib/cluster/cluster-slots.ts b/packages/client/lib/cluster/cluster-slots.ts index f6a4789a1a3..ed7e7b8497a 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -137,6 +137,32 @@ export function groupCommandsByDestination< return { byDestination, unrouted }; } +/** + * Splits commands extracted from a dying node's queue into the queued tail of + * a chain (MULTI/pipeline) whose head is already sent (out of view, in + * `#waitingForReply`) and everything else, which is safe to relocate as-is. + * + * Only used on full node loss: the connection is about to be destroyed, so + * the in-flight chain's tail is rejected rather than relocated - relocating a + * fragment of an already-partially-sent chain would run it out of order or + * split it across two connections. This doesn't apply to partial slot + * migration, where the source connection survives and keeps its in-flight + * chain intact. + */ +export function splitInFlightChainTail( + commands: CommandToWrite[], + chainInExecution: symbol | undefined +) { + const inFlightChainTail = chainInExecution === undefined + ? [] + : commands.filter(command => command.chainId === chainInExecution); + const relocatable = inFlightChainTail.length === 0 + ? commands + : commands.filter(command => command.chainId !== chainInExecution); + + return { inFlightChainTail, relocatable }; +} + export type OnShardedChannelMovedError = ( err: unknown, channel: string, @@ -492,12 +518,7 @@ export default class RedisClusterSlots< // relocated on its own - it would run as a fragment of the // transaction on a different node than the part already sent. const chainInExecution = sourceNode.client._getQueue().chainInExecution; - const inFlightChainTail = chainInExecution === undefined - ? [] - : remainingCommands.filter(command => command.chainId === chainInExecution); - const relocatable = inFlightChainTail.length === 0 - ? remainingCommands - : remainingCommands.filter(command => command.chainId !== chainInExecution); + const { inFlightChainTail, relocatable } = splitInFlightChainTail(remainingCommands, chainInExecution); if (inFlightChainTail.length > 0) { sourceNode.client._getQueue().rejectCommands(inFlightChainTail, new DisconnectsClientError());