Hitless upgrades cluster

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

@@ -577,11 +577,12 @@ export default class RedisCommandsQueue {
   /**
   * Gets all commands from the write queue without removing them.
   */
-  getAllCommands(): CommandToWrite[] {
+  extractAllCommands(): CommandToWrite[] {
     const result: CommandToWrite[] = [];
     let current = this.#toWrite.head;
     while(current) {
       result.push(current.value);
+      this.#toWrite.remove(current);
       current = current.next;
     }
     return result;
@@ -591,7 +592,11 @@ export default class RedisCommandsQueue {
    * Prepends commands to the write queue in reverse.
    */
   prependCommandsToWrite(commands: CommandToWrite[]) {
-    for (let i = commands.length - 1; i <= 0; i--) {
+    if (!commands.length) {
+      return;
+    }
+
+    for (let i = commands.length - 1; i >= 0; i--) {
       this.#toWrite.unshift(commands[i]);
     }
   }

packages/client/lib/client/index.ts

@@ -1018,7 +1018,7 @@ export default class RedisClient<
     * @internal
     */
    _getQueue(): RedisCommandsQueue {
-     return this.#queue;
+     return this._self.#queue;
    }
 
   /**

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

@@ -356,7 +356,7 @@ export default class RedisClusterSlots<
       } else {
         // If source shard doesnt have any slots left, this means we can safely move all commands to the new shard.
         // Same goes for sharded pub sub listeners
-        const normalCommandsToMove = sourceNode.client!._getQueue().getAllCommands();
+        const normalCommandsToMove = sourceNode.client!._getQueue().extractAllCommands();
         // 5. Prepend extracted commands, chans
         destMasterNode.client?._getQueue().prependCommandsToWrite(normalCommandsToMove);
         if('pubSub' in sourceNode) {
@@ -371,9 +371,13 @@ export default class RedisClusterSlots<
         this.nodeByAddress.delete(sourceAddress);
 
         // 4.3 Kill because no slots are pointing to it anymore
-        await sourceNode.client?.close()
+        if (sourceNode.client?.isOpen) {
+          await sourceNode.client?.close()
+        }
         if('pubSub' in sourceNode) {
-          await sourceNode.pubSub?.client.close();
+          if (sourceNode.pubSub?.client.isOpen) {
+            await sourceNode.pubSub?.client.close();
+          }
         }
       }