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/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, CommandTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, SocketClosedUnexpectedlyError, WatchError } from '../errors';
 import { defineScript } from '../lua-script';
 import { spy } from 'sinon';
 import { once } from 'node:events';
@@ -263,8 +263,41 @@ describe('Client', () => {
           AbortError
         );
       }, GLOBAL.SERVERS.OPEN);
+
+      testUtils.testWithClient('AbortError with timeout', client => {
+        const controller = new AbortController();
+        controller.abort();
+
+        return assert.rejects(
+          client.sendCommand(['PING'], {
+            abortSignal: controller.signal
+          }),
+          AbortError
+        );
+      }, {
+        ...GLOBAL.SERVERS.OPEN,
+        clientOptions: {
+          commandTimeout: 50,
+        }
+      });
     });
 
+  testUtils.testWithClient('CommandTimeoutError', async client => {
+    const promise = assert.rejects(client.sendCommand(['PING']), CommandTimeoutError),
+      start = process.hrtime.bigint();
+
+    while (process.hrtime.bigint() - start < 50_000_000) {
+      // block the event loop for 1ms, to make sure the connection will timeout
+    }
+
+    await promise;
+  }, {
+    ...GLOBAL.SERVERS.OPEN,
+    clientOptions: {
+      commandTimeout: 50,
+    }
+  });
+
     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

@@ -4,7 +4,7 @@ import { BasicAuth, CredentialsError, CredentialsProvider, StreamingCredentialsP
 import RedisCommandsQueue, { CommandOptions } from './commands-queue';
 import { EventEmitter } from 'node:events';
 import { attachConfig, functionArgumentsPrefix, getTransformReply, scriptArgumentsPrefix } from '../commander';
-import { ClientClosedError, ClientOfflineError, DisconnectsClientError, WatchError } from '../errors';
+import { ClientClosedError, ClientOfflineError, CommandTimeoutError, DisconnectsClientError, WatchError } from '../errors';
 import { URL } from 'node:url';
 import { TcpSocketConnectOpts } from 'node:net';
 import { PUBSUB_TYPE, PubSubType, PubSubListener, PubSubTypeListeners, ChannelListeners } from './pub-sub';
@@ -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<
@@ -889,9 +893,41 @@ export default class RedisClient<
       return Promise.reject(new ClientOfflineError());
     }
 
+    let controller: AbortController;
+    if (this._self.#options?.commandTimeout) {
+      controller = new AbortController()
+      let abortSignal = controller.signal;
+      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;
+    if (!this._self.#options?.commandTimeout) {
+      return promise;
+    }
+
+    return new Promise<T>((resolve, reject) => {
+      const timeoutId = setTimeout(() => {
+        controller.abort();
+        reject(new CommandTimeoutError());
+      }, this._self.#options?.commandTimeout)
+      promise.then(result => {
+        clearInterval(timeoutId);
+        resolve(result)
+      }).catch(error => {
+        clearInterval(timeoutId);
+        reject(error)
+      });
+    })
   }
 
   async SELECT(db: number): Promise<void> {

packages/client/lib/errors.ts

@@ -22,6 +22,12 @@ export class SocketTimeoutError extends Error {
   }
 }
 
+export class CommandTimeoutError extends Error {
+  constructor() {
+    super('Command timeout');
+  }
+}
+
 export class ClientClosedError extends Error {
   constructor() {
     super('The client is closed');