From 3192efac0c44ac12b5effc21ab3973ea555ffacc 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 89c467c8d6e..d07025e1fc5 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -151,6 +151,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 ff168a88b4b..2eaef18fd2f 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -1853,7 +1853,8 @@ export default class RedisClient< */ async _executePipeline( commands: Array, - selectedDB?: number + selectedDB?: number, + slotNumber?: number ) { assertNoHimportSessionCommands(commands); @@ -1871,7 +1872,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), @@ -1911,7 +1913,8 @@ export default class RedisClient< */ async _executeMulti( commands: Array, - selectedDB?: number + selectedDB?: number, + slotNumber?: number ) { assertNoHimportSessionCommands(commands); @@ -1939,20 +1942,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 38d8e2e774d..2745cc83260 100644 --- a/packages/client/lib/cluster/cluster-slots.spec.ts +++ b/packages/client/lib/cluster/cluster-slots.spec.ts @@ -1,10 +1,31 @@ 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'; import { ClientClosedError } from '../errors'; 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; @@ -57,4 +78,44 @@ describe('RedisClusterSlots', () => { assert.throws(() => slots.getRandomNode(), ClientClosedError) }); }); + + 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 5f82fd1189e..1cabf93fb3a 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'; @@ -136,6 +137,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, @@ -449,12 +484,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 @@ -494,17 +551,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 5467381aad0..89932504f88 100644 --- a/packages/client/lib/cluster/index.ts +++ b/packages/client/lib/cluster/index.ts @@ -808,12 +808,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 f6abaeae4f3284ec1d0d5bc3140af75b002b328b 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 2745cc83260..2e7e55b8dbe 100644 --- a/packages/client/lib/cluster/cluster-slots.spec.ts +++ b/packages/client/lib/cluster/cluster-slots.spec.ts @@ -98,24 +98,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 1cabf93fb3a..efcadc94b41 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -152,13 +152,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) { @@ -168,7 +176,7 @@ export function groupCommandsByDestination< } } - return byDestination; + return { byDestination, unrouted }; } export type OnShardedChannelMovedError = ( @@ -483,34 +491,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`); } } @@ -573,7 +569,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 b1acc6d5ad2d9b9bdd0968597552e8cdf3554b8e 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 d07025e1fc5..d7caf56e301 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -710,6 +710,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[] = []; @@ -717,7 +726,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 4b92ce1fa6077c4c89ab49bcc7158e47def42b69 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 d7caf56e301..fd9f6fb813b 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -713,37 +713,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 efcadc94b41..f4159adccc7 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -492,13 +492,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 c090c86955839a9c11e3c36a11756402b855b700 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 fd9f6fb813b..537924e875d 100644 --- a/packages/client/lib/client/commands-queue.ts +++ b/packages/client/lib/client/commands-queue.ts @@ -719,8 +719,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 9f62190d2453ae794b3148205fc59245cd94b022 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 2e7e55b8dbe..700d9bac22b 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'; import { ClientClosedError } from '../errors'; @@ -134,4 +134,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 f4159adccc7..3fa09ca0e8e 100644 --- a/packages/client/lib/cluster/cluster-slots.ts +++ b/packages/client/lib/cluster/cluster-slots.ts @@ -179,6 +179,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, @@ -556,12 +582,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());