feat: Add command timeout

docs/client-configuration.md

@@ -29,6 +29,7 @@
 | isolationPoolOptions         |                                          | An object that configures a pool of isolated connections, If you frequently need isolated connections, consider using [createClientPool](https://github.com/redis/node-redis/blob/master/docs/pool.md#creating-a-pool) instead                                                                                                     |
 | pingInterval                 |                                          | Send `PING` command at interval (in ms). Useful with ["Azure Cache for Redis"](https://learn.microsoft.com/en-us/azure/azure-cache-for-redis/cache-best-practices-connection#idle-timeout)                                                          |
 | disableClientInfo            | `false`                                  | Disables `CLIENT SETINFO LIB-NAME node-redis` and `CLIENT SETINFO LIB-VER X.X.X` commands                                                                                                                                                           | 
+| commandTimeout               |                                          | Throw an error and abort a command if it takes longer than the specified time (in milliseconds).                                                                                                                                                    |
 
 ## Reconnect Strategy
 

package.json

@@ -23,7 +23,7 @@
     "@istanbuljs/nyc-config-typescript": "^1.0.2",
     "@release-it/bumper": "^7.0.5",
     "@types/mocha": "^10.0.6",
-    "@types/node": "^20.11.16",
+    "@types/node": "^20.19.1",
     "gh-pages": "^6.1.1",
     "mocha": "^10.2.0",
     "nyc": "^15.1.0",

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

@@ -144,7 +144,7 @@ export default class RedisCommandsQueue {
     if (this.#maxLength && this.#toWrite.length + this.#waitingForReply.length >= this.#maxLength) {
       return Promise.reject(new Error('The queue is full'));
     } else if (options?.abortSignal?.aborted) {
-      return Promise.reject(new AbortError());
+      return Promise.reject(new AbortError(options?.abortSignal?.reason));
     }
 
     return new Promise((resolve, reject) => {
@@ -165,7 +165,7 @@ export default class RedisCommandsQueue {
           signal,
           listener: () => {
             this.#toWrite.remove(node);
-            value.reject(new AbortError());
+            value.reject(new AbortError(signal.reason));
           }
         };
         signal.addEventListener('abort', value.abort.listener, { once: true });

packages/client/lib/client/index.spec.ts

@@ -1,7 +1,7 @@
 import { strict as assert } from 'node:assert';
 import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
 import RedisClient, { RedisClientOptions, RedisClientType } from '.';
-import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, SocketClosedUnexpectedlyError, WatchError } from '../errors';
+import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, WatchError } from '../errors';
 import { defineScript } from '../lua-script';
 import { spy } from 'sinon';
 import { once } from 'node:events';
@@ -263,8 +263,47 @@ describe('Client', () => {
           AbortError
         );
       }, GLOBAL.SERVERS.OPEN);
+
+      testUtils.testWithClient('rejects with AbortError - respects given abortSignal', client => {
+
+        const promise = client.sendCommand(['PING'], {
+          abortSignal: AbortSignal.abort("my reason")
+        })
+
+        assert.rejects(
+          promise,
+          AbortError
+        );
+
+        promise.catch((error: unknown) => {
+          assert.ok((error as string).includes("my reason"));
+        });
+
+      }, {
+        ...GLOBAL.SERVERS.OPEN,
+        clientOptions: {
+          commandTimeout: 50,
+        }
+      });
     });
 
+
+  testUtils.testWithClient('rejects with AbortError on commandTimeout timer', async client => {
+    const start = process.hrtime.bigint();
+    const promise = client.ping();
+
+    while (process.hrtime.bigint() - start < 10_000_000) {
+      // block the event loop for 10ms, to make sure the connection will timeout
+    };
+
+    assert.rejects(promise, AbortError);
+  }, {
+    ...GLOBAL.SERVERS.OPEN,
+    clientOptions: {
+      commandTimeout: 10,
+    }
+  });
+
     testUtils.testWithClient('undefined and null should not break the client', async client => {
       await assert.rejects(
         client.sendCommand([null as any, undefined as any]),

packages/client/lib/client/index.ts

@@ -144,6 +144,10 @@ export interface RedisClientOptions<
    * Tag to append to library name that is sent to the Redis server
    */
   clientInfoTag?: string;
+  /**
+   * Provides a timeout in milliseconds.
+   */
+  commandTimeout?: number;
 }
 
 type WithCommands<
@@ -526,7 +530,7 @@ export default class RedisClient<
   async #handshake(chainId: symbol, asap: boolean) {
     const promises = [];
     const commandsWithErrorHandlers = await this.#getHandshakeCommands();
-    
+
     if (asap) commandsWithErrorHandlers.reverse()
 
     for (const { cmd, errorHandler } of commandsWithErrorHandlers) {
@@ -632,7 +636,7 @@ export default class RedisClient<
           // since they could be connected to an older version that doesn't support them.
         }
       });
-      
+
       commands.push({
         cmd: [
           'CLIENT',
@@ -889,7 +893,21 @@ export default class RedisClient<
       return Promise.reject(new ClientOfflineError());
     }
 
+    if (this._self.#options?.commandTimeout) {
+      let abortSignal = AbortSignal.timeout(this._self.#options?.commandTimeout);
+      if (options?.abortSignal) {
+        abortSignal = AbortSignal.any([
+          abortSignal,
+       	  options.abortSignal
+       	]);
+      }
+      options = {
+        ...options,
+        abortSignal
+      }
+    }
     const promise = this._self.#queue.addCommand<T>(args, options);
+
     this._self.#scheduleWrite();
     return promise;
   }

packages/client/lib/errors.ts

@@ -1,6 +1,6 @@
 export class AbortError extends Error {
-  constructor() {
-    super('The command was aborted');
+  constructor(message = '') {
+    super(`The command was aborted: ${message}`);
   }
 }