Hitless upgrades (#3021)

* feat(errors): Add specialized timeout error types for maintenance scenarios

- Added `SocketTimeoutDuringMaintananceError`, a subclass of `TimeoutError`, to handle socket timeouts during maintenance.
- Added `CommandTimeoutDuringMaintenanceError`, another subclass of `TimeoutError`, to address command write timeouts during maintenance.

* feat(linked-list): Add EmptyAwareSinglyLinkedList and enhance DoublyLinkedList functionality

- Introduced `EmptyAwareSinglyLinkedList`, a subclass of `SinglyLinkedList` that emits an `empty` event when the list becomes empty due to `reset`, `shift`, or `remove` operations.
- Added `nodes()` iterator method to `DoublyLinkedList` for iterating over nodes directly.
- Enhanced unit tests for `DoublyLinkedList` and `SinglyLinkedList` to cover edge cases and new functionality.
- Added comprehensive tests for `EmptyAwareSinglyLinkedList` to validate `empty` event emission under various scenarios.
- Improved code formatting and consistency.

* refactor(commands-queue): Improve push notification handling

- Replaced `setInvalidateCallback` with a more flexible `addPushHandler` method, allowing multiple handlers for push notifications.
- Introduced the `PushHandler` type to standardize push notification processing.
- Refactored `RedisCommandsQueue` to use a `#pushHandlers` array, enabling dynamic and modular handling of push notifications.
- Updated `RedisClient` to leverage the new handler mechanism for `invalidate` push notifications, simplifying and decoupling logic.

* feat(commands-queue): Add method to wait for in-flight commands to complete

- Introduced `waitForInflightCommandsToComplete` method to asynchronously wait for all in-flight commands to finish processing.
- Utilized the `empty` event from `#waitingForReply` to signal when all commands have been completed.

* feat(commands-queue): Introduce maintenance mode support for commands-queue

- Added `#maintenanceCommandTimeout` and `setMaintenanceCommandTimeout` method to dynamically adjust command timeouts during maintenance

* refator(client): Extract socket event listener setup into helper method

* refactor(socket): Add maintenance mode support and dynamic timeout handling

- Added `#maintenanceTimeout` and `setMaintenanceTimeout` method to dynamically adjust socket timeouts during maintenance.

* feat(client): Add Redis Enterprise maintenance configuration options

- Added `maintPushNotifications` option to control how the client handles Redis Enterprise maintenance push notifications (`disabled`, `enabled`, `au
to`).
- Added `maintMovingEndpointType` option to specify the endpoint type for reconnecting during a MOVING notification (`auto`, `internal-ip`, `external-ip`,
etc.).
- Added `maintRelaxedCommandTimeout` option to define a relaxed timeout for commands during maintenance.
- Added `maintRelaxedSocketTimeout` option to define a relaxed timeout for the socket during maintenance.
- Enforced RESP3 requirement for maintenance-related features (`maintPushNotifications`).

* feat(client): Add socket helpers and pause mechanism

- Introduced `#paused` flag with corresponding `_pause` and `_unpause` methods to
temporarily halt writing commands to the socket during maintenance windows.
- Updated `#write` method to respect the `#paused` flag, preventing new commands from being written during maintenance.
- Added `_ejectSocket` method to safely detach from and return the current socket
- Added `_insertSocket` method to receive and start using a new socket

* feat(client): Add Redis Enterprise maintenance handling capabilities

- Introduced `EnterpriseMaintenanceManager` to manage Redis Enterprise maintenance events and push notifications.
- Integrated `EnterpriseMaintenanceManager` into `RedisClient` to handle maintenance push notifications and manage socket transitions.
- Implemented graceful handling of MOVING, MIGRATING, and FAILOVER push notifications, including socket replacement and timeout adjustments.


* test: add E2E test infrastructure for Redis maintenance scenarios

* test: add E2E tests for Redis Enterprise maintenance timeout handling (#3)

* test: add connection handoff test


---------

Co-authored-by: Pavel Pashov <[email protected]>
Co-authored-by: Pavel Pashov <[email protected]>

Author: nkaradzhov

package-lock.json

@@ -13,7 +13,7 @@
     "release": "release-it"
   },
   "peerDependencies": {
-    "@redis/client": "^5.8.2"
+    "@redis/client": "^5.8.2 || ^5.9.0-0"
   },
   "devDependencies": {
     "@redis/test-utils": "*"

packages/bloom/package.json

@@ -1,10 +1,11 @@
-import { SinglyLinkedList, DoublyLinkedNode, DoublyLinkedList } from './linked-list';
+import { DoublyLinkedNode, DoublyLinkedList, EmptyAwareSinglyLinkedList } from './linked-list';
 import encodeCommand from '../RESP/encoder';
 import { Decoder, PUSH_TYPE_MAPPING, RESP_TYPES } from '../RESP/decoder';
 import { TypeMapping, ReplyUnion, RespVersions, RedisArgument } from '../RESP/types';
 import { ChannelListeners, PubSub, PubSubCommand, PubSubListener, PubSubType, PubSubTypeListeners } from './pub-sub';
-import { AbortError, ErrorReply, TimeoutError } from '../errors';
+import { AbortError, ErrorReply, CommandTimeoutDuringMaintenanceError, TimeoutError } from '../errors';
 import { MonitorCallback } from '.';
+import { dbgMaintenance } from './enterprise-maintenance-manager';
 
 export interface CommandOptions<T = TypeMapping> {
   chainId?: symbol;
@@ -30,6 +31,7 @@ export interface CommandToWrite extends CommandWaitingForReply {
   timeout: {
     signal: AbortSignal;
     listener: () => unknown;
+    originalTimeout: number | undefined;
   } | undefined;
 }
 
@@ -50,22 +52,74 @@ const RESP2_PUSH_TYPE_MAPPING = {
   [RESP_TYPES.SIMPLE_STRING]: Buffer
 };
 
+// Try to handle a push notification. Return whether you
+// successfully consumed the notification or not. This is
+// important in order for the queue to be able to pass the
+// notification to another handler if the current one did not
+// succeed.
+type PushHandler = (pushItems: Array<any>) => boolean;
+
 export default class RedisCommandsQueue {
   readonly #respVersion;
   readonly #maxLength;
   readonly #toWrite = new DoublyLinkedList<CommandToWrite>();
-  readonly #waitingForReply = new SinglyLinkedList<CommandWaitingForReply>();
+  readonly #waitingForReply = new EmptyAwareSinglyLinkedList<CommandWaitingForReply>();
   readonly #onShardedChannelMoved;
   #chainInExecution: symbol | undefined;
   readonly decoder;
   readonly #pubSub = new PubSub();
 
+  #pushHandlers: PushHandler[] = [this.#onPush.bind(this)];
+
+  #maintenanceCommandTimeout: number | undefined
+
+  setMaintenanceCommandTimeout(ms: number | undefined) {
+    // Prevent possible api misuse
+    if (this.#maintenanceCommandTimeout === ms) {
+      dbgMaintenance(`Queue already set maintenanceCommandTimeout to ${ms}, skipping`);
+      return;
+    };
+
+    dbgMaintenance(`Setting maintenance command timeout to ${ms}`);
+    this.#maintenanceCommandTimeout = ms;
+
+    if(this.#maintenanceCommandTimeout === undefined) {
+      dbgMaintenance(`Queue will keep maintenanceCommandTimeout for exisitng commands, just to be on the safe side. New commands will receive normal timeouts`);
+      return;
+    }
+
+    let counter = 0;
+    const total = this.#toWrite.length;
+
+    // Overwrite timeouts of all eligible toWrite commands
+    for(const node of this.#toWrite.nodes()) {
+      const command = node.value;
+
+      // Remove timeout listener if it exists
+      RedisCommandsQueue.#removeTimeoutListener(command)
+
+      counter++;
+      const newTimeout = this.#maintenanceCommandTimeout;
+
+      // Overwrite the command's timeout
+      const signal = AbortSignal.timeout(newTimeout);
+      command.timeout = {
+        signal,
+        listener: () => {
+          this.#toWrite.remove(node);
+          command.reject(new CommandTimeoutDuringMaintenanceError(newTimeout));
+        },
+        originalTimeout: command.timeout?.originalTimeout
+      };
+      signal.addEventListener('abort', command.timeout.listener, { once: true });
+    };
+    dbgMaintenance(`Total of ${counter} of ${total} timeouts reset to ${ms}`);
+  }
+
   get isPubSubActive() {
     return this.#pubSub.isActive;
   }
 
-  #invalidateCallback?: (key: RedisArgument | null) => unknown;
-
   constructor(
     respVersion: RespVersions,
     maxLength: number | null | undefined,
@@ -107,6 +161,7 @@ export default class RedisCommandsQueue {
       }
       return true;
     }
+    return false
   }
 
   #getTypeMapping() {
@@ -119,30 +174,27 @@ export default class RedisCommandsQueue {
       onErrorReply: err => this.#onErrorReply(err),
       //TODO: we can shave off a few cycles by not adding onPush handler at all if CSC is not used
       onPush: push => {
-        if (!this.#onPush(push)) {
-          // currently only supporting "invalidate" over RESP3 push messages
-          switch (push[0].toString()) {
-            case "invalidate": {
-              if (this.#invalidateCallback) {
-                if (push[1] !== null) {
-                  for (const key of push[1]) {
-                    this.#invalidateCallback(key);
-                  }
-                } else {
-                  this.#invalidateCallback(null);
-                }
-              }
-              break;
-            }
-          }
+        for(const pushHandler of this.#pushHandlers) {
+          if(pushHandler(push)) return
         }
       },
       getTypeMapping: () => this.#getTypeMapping()
     });
   }
 
-  setInvalidateCallback(callback?: (key: RedisArgument | null) => unknown) {
-    this.#invalidateCallback = callback;
+  addPushHandler(handler: PushHandler): void {
+    this.#pushHandlers.push(handler);
+  }
+
+  async waitForInflightCommandsToComplete(): Promise<void> {
+    // In-flight commands already completed
+    if(this.#waitingForReply.length === 0) {
+      return
+    };
+    // Otherwise wait for in-flight commands to fire `empty` event
+    return new Promise(resolve => {
+      this.#waitingForReply.events.on('empty', resolve)
+    });
   }
 
   addCommand<T>(
@@ -168,15 +220,20 @@ export default class RedisCommandsQueue {
         typeMapping: options?.typeMapping
       };
 
-      const timeout = options?.timeout;
+      // If #maintenanceCommandTimeout was explicitly set, we should
+      // use it instead of the timeout provided by the command
+      const timeout = this.#maintenanceCommandTimeout ?? options?.timeout;
+      const wasInMaintenance = this.#maintenanceCommandTimeout !== undefined;
       if (timeout) {
+
         const signal = AbortSignal.timeout(timeout);
         value.timeout = {
           signal,
           listener: () => {
             this.#toWrite.remove(node);
-            value.reject(new TimeoutError());
-          }
+            value.reject(wasInMaintenance ? new CommandTimeoutDuringMaintenanceError(timeout) : new TimeoutError());
+          },
+          originalTimeout: options?.timeout
         };
         signal.addEventListener('abort', value.timeout.listener, { once: true });
       }
@@ -432,7 +489,7 @@ export default class RedisCommandsQueue {
   }
 
   static #removeTimeoutListener(command: CommandToWrite) {
-    command.timeout!.signal.removeEventListener('abort', command.timeout!.listener);
+    command.timeout?.signal.removeEventListener('abort', command.timeout!.listener);
   }
 
   static #flushToWrite(toBeSent: CommandToWrite, err: Error) {