add command timeout

Author: florian-schunk

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';
@@ -231,6 +231,22 @@ describe('Client', () => {
       }, GLOBAL.SERVERS.OPEN);
     });
 
+  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';
@@ -80,6 +80,10 @@ export interface RedisClientOptions<
    * TODO
    */
   commandOptions?: CommandOptions<TYPE_MAPPING>;
+  /**
+   * Provides a timeout in milliseconds.
+   */
+  commandTimeout?: number;
 }
 
 type WithCommands<
@@ -730,9 +734,34 @@ export default class RedisClient<
       return Promise.reject(new ClientOfflineError());
     }
 
+    let controller: AbortController;
+    if (this._self.#options?.commandTimeout) {
+      controller = new AbortController()
+      options = {
+        ...options,
+        abortSignal: controller.signal
+      }
+    }
     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

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