feat: Мove creation of redis client outside of cache service (#4409)

Signed-off-by: nikolay <n.atanasow94@gmail.com>
Signed-off-by: Konstantina Blazhukova <konstantina.blajukova@gmail.com>
Co-authored-by: Nikolay Atanasow <n.atanasow94@gmail.com>

Author: konstantinabl

package-lock.json

@@ -1,10 +1,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
-import type { ICacheClient } from './ICacheClient';
 import { RequestDetails } from '../../types';
+import type { ICacheClient } from './ICacheClient';
 
 export interface IRedisCacheClient extends ICacheClient {
-  disconnect: () => Promise<void>;
   incrBy(key: string, amount: number, callingMethod: string, requestDetails: RequestDetails): Promise<number>;
   rPush(key: string, value: any, callingMethod: string, requestDetails: RequestDetails): Promise<number>;
   lRange(

packages/relay/src/lib/clients/cache/IRedisCacheClient.ts

@@ -3,10 +3,9 @@
 import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
 import { Logger } from 'pino';
 import { Registry } from 'prom-client';
-import { createClient, RedisClientType } from 'redis';
+import { RedisClientType } from 'redis';
 
 import { Utils } from '../../../utils';
-import { RedisCacheError } from '../../errors/RedisCacheError';
 import { IRedisCacheClient } from './IRedisCacheClient';
 
 /**
@@ -41,66 +40,16 @@ export class RedisCache implements IRedisCacheClient {
    */
   private readonly client: RedisClientType;
 
-  /**
-   * Boolean showing if the connection to the Redis client has finished.
-   * @private
-   */
-  private connected: Promise<boolean>;
-
   /**
    * Creates an instance of `RedisCache`.
    *
    * @param {Logger} logger - The logger instance.
    * @param {Registry} register - The metrics registry.
    */
-  public constructor(logger: Logger, register: Registry) {
+  public constructor(logger: Logger, register: Registry, client: RedisClientType) {
     this.logger = logger;
     this.register = register;
-
-    const redisUrl = ConfigService.get('REDIS_URL')!;
-    const reconnectDelay = ConfigService.get('REDIS_RECONNECT_DELAY_MS');
-    this.client = createClient({
-      url: redisUrl,
-      socket: {
-        reconnectStrategy: (retries: number) => {
-          const delay = retries * reconnectDelay;
-          logger.warn(`Trying to reconnect with Redis, retry #${retries}. Delay is ${delay} ms...`);
-          return delay;
-        },
-      },
-    });
-    this.connected = this.client
-      .connect()
-      .then(() => true)
-      .catch((error) => {
-        this.logger.error(error, 'Redis connection could not be established!');
-        return false;
-      });
-    this.client.on('ready', async () => {
-      this.connected = Promise.resolve(true);
-      const connections = await this.getNumberOfConnections().catch((error) => {
-        this.logger.error(error);
-        return 0;
-      });
-      logger.info(`Connected to Redis server (${redisUrl}) successfully! Number of connections: ${connections}`);
-    });
-    this.client.on('end', () => {
-      this.connected = Promise.resolve(false);
-      logger.info('Disconnected from Redis server!');
-    });
-    this.client.on('error', (error) => {
-      this.connected = Promise.resolve(false);
-      const redisError = new RedisCacheError(error);
-      if (redisError.isSocketClosed()) {
-        logger.error(`Error occurred with Redis Connection when closing socket: ${redisError.message}`);
-      } else {
-        logger.error(`Error occurred with Redis Connection: ${redisError.fullError}`);
-      }
-    });
-  }
-
-  async getConnectedClient(): Promise<RedisClientType> {
-    return this.isConnected().then(() => this.client);
+    this.client = client;
   }
 
   /**
@@ -111,8 +60,7 @@ export class RedisCache implements IRedisCacheClient {
    * @returns The cached value or null if not found.
    */
   async get(key: string, callingMethod: string): Promise<any> {
-    const client = await this.getConnectedClient();
-    const result = await client.get(key);
+    const result = await this.client.get(key);
     if (result) {
       if (this.logger.isLevelEnabled('trace')) {
         const censoredKey = key.replace(Utils.IP_ADDRESS_REGEX, '<REDACTED>');
@@ -135,13 +83,12 @@ export class RedisCache implements IRedisCacheClient {
    * @returns A Promise that resolves when the value is cached.
    */
   async set(key: string, value: any, callingMethod: string, ttl?: number): Promise<void> {
-    const client = await this.getConnectedClient();
     const serializedValue = JSON.stringify(value);
     const resolvedTtl = ttl ?? this.options.ttl; // in milliseconds
     if (resolvedTtl > 0) {
-      await client.set(key, serializedValue, { PX: resolvedTtl });
+      await this.client.set(key, serializedValue, { PX: resolvedTtl });
     } else {
-      await client.set(key, serializedValue);
+      await this.client.set(key, serializedValue);
     }
 
     const censoredKey = key.replace(Utils.IP_ADDRESS_REGEX, '<REDACTED>');
@@ -163,15 +110,14 @@ export class RedisCache implements IRedisCacheClient {
    * @returns A Promise that resolves when the values are cached.
    */
   async multiSet(keyValuePairs: Record<string, any>, callingMethod: string): Promise<void> {
-    const client = await this.getConnectedClient();
     // Serialize values
     const serializedKeyValuePairs: Record<string, string> = {};
     for (const [key, value] of Object.entries(keyValuePairs)) {
       serializedKeyValuePairs[key] = JSON.stringify(value);
     }
 
     // Perform mSet operation
-    await client.mSet(serializedKeyValuePairs);
+    await this.client.mSet(serializedKeyValuePairs);
 
     // Log the operation
     const entriesLength = Object.keys(keyValuePairs).length;
@@ -189,10 +135,9 @@ export class RedisCache implements IRedisCacheClient {
    * @returns A Promise that resolves when the values are cached.
    */
   async pipelineSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number): Promise<void> {
-    const client = await this.getConnectedClient();
     const resolvedTtl = ttl ?? this.options.ttl; // in milliseconds
 
-    const pipeline = client.multi();
+    const pipeline = this.client.multi();
 
     for (const [key, value] of Object.entries(keyValuePairs)) {
       const serializedValue = JSON.stringify(value);
@@ -217,66 +162,13 @@ export class RedisCache implements IRedisCacheClient {
    * @returns A Promise that resolves when the value is deleted from the cache.
    */
   async delete(key: string, callingMethod: string): Promise<void> {
-    const client = await this.getConnectedClient();
-    await client.del(key);
+    await this.client.del(key);
     if (this.logger.isLevelEnabled('trace')) {
       this.logger.trace(`delete cache for ${key} on ${callingMethod} call`);
     }
     // TODO: add metrics
   }
 
-  /**
-   * Clears the entire cache.
-   *
-   * @returns {Promise<void>} A Promise that resolves when the cache is cleared.
-   */
-  async clear(): Promise<void> {
-    const client = await this.getConnectedClient();
-    await client.flushAll();
-  }
-
-  /**
-   * Checks if the client is connected to the Redis server.
-   *
-   * @returns {Promise<boolean>} A Promise that resolves to true if the client is connected, false otherwise.
-   */
-  async isConnected(): Promise<boolean> {
-    return this.connected;
-  }
-
-  /**
-   * Retrieves the number of connections to the Redis server.
-   *
-   * @returns {Promise<number>} A Promise that resolves to the number of connections.
-   * @throws {Error} If an error occurs while retrieving the number of connections.
-   */
-  async getNumberOfConnections(): Promise<number> {
-    const client = await this.getConnectedClient();
-    const clientList = await client.clientList();
-    return clientList.length;
-  }
-
-  /**
-   * Connects the client to the Redis server.
-   *
-   * @returns {Promise<void>} A Promise that resolves when the client is connected.
-   * @throws {Error} If an error occurs while connecting to Redis.
-   */
-  async connect(): Promise<void> {
-    await this.client.connect();
-  }
-
-  /**
-   * Disconnects the client from the Redis server.
-   *
-   * @returns {Promise<void>} A Promise that resolves when the client is disconnected.
-   * @throws {Error} If an error occurs while disconnecting from Redis.
-   */
-  async disconnect(): Promise<void> {
-    const client = await this.getConnectedClient();
-    await client.quit();
-  }
-
   /**
    * Increments a value in the cache.
    *
@@ -286,8 +178,7 @@ export class RedisCache implements IRedisCacheClient {
    * @returns The value of the key after incrementing
    */
   async incrBy(key: string, amount: number, callingMethod: string): Promise<number> {
-    const client = await this.getConnectedClient();
-    const result = await client.incrBy(key, amount);
+    const result = await this.client.incrBy(key, amount);
     if (this.logger.isLevelEnabled('trace')) {
       this.logger.trace(`incrementing ${key} by ${amount} on ${callingMethod} call`);
     }
@@ -304,8 +195,7 @@ export class RedisCache implements IRedisCacheClient {
    * @returns The list of elements in the range
    */
   async lRange(key: string, start: number, end: number, callingMethod: string): Promise<any[]> {
-    const client = await this.getConnectedClient();
-    const result = await client.lRange(key, start, end);
+    const result = await this.client.lRange(key, start, end);
     if (this.logger.isLevelEnabled('trace')) {
       this.logger.trace(`retrieving range [${start}:${end}] from ${key} on ${callingMethod} call`);
     }
@@ -321,9 +211,8 @@ export class RedisCache implements IRedisCacheClient {
    * @returns The length of the list after pushing
    */
   async rPush(key: string, value: any, callingMethod: string): Promise<number> {
-    const client = await this.getConnectedClient();
     const serializedValue = JSON.stringify(value);
-    const result = await client.rPush(key, serializedValue);
+    const result = await this.client.rPush(key, serializedValue);
     if (this.logger.isLevelEnabled('trace')) {
       this.logger.trace(`pushing ${serializedValue} to ${key} on ${callingMethod} call`);
     }
@@ -337,11 +226,19 @@ export class RedisCache implements IRedisCacheClient {
    * @returns The list of keys matching the pattern
    */
   async keys(pattern: string, callingMethod: string): Promise<string[]> {
-    const client = await this.getConnectedClient();
-    const result = await client.keys(pattern);
+    const result = await this.client.keys(pattern);
     if (this.logger.isLevelEnabled('trace')) {
       this.logger.trace(`retrieving keys matching ${pattern} on ${callingMethod} call`);
     }
     return result;
   }
+
+  /**
+   * Clears the entire cache.
+   *
+   * @returns {Promise<void>} A Promise that resolves when the cache is cleared.
+   */
+  async clear(): Promise<void> {
+    await this.client.flushAll();
+  }
 }

packages/relay/src/lib/clients/cache/redisCache.ts

@@ -0,0 +1,59 @@
+// SPDX-License-Identifier: Apache-2.0
+import { Logger } from 'pino';
+import { createClient, RedisClientType } from 'redis';
+
+import { RedisCacheError } from '../errors/RedisCacheError';
+
+export class RedisClientManager {
+  private client: RedisClientType;
+
+  private connected: boolean = false;
+
+  constructor(
+    private readonly logger: Logger,
+    url: string,
+    reconnectMs: number,
+  ) {
+    this.client = createClient({
+      url,
+      socket: { reconnectStrategy: (retries) => retries * reconnectMs },
+    });
+    this.client.on('ready', () => {
+      this.logger.info(`Redis client connected to ${url}`);
+    });
+    this.client.on('end', () => {
+      this.logger.info('Disconnected from Redis server!');
+    });
+    this.client.on('error', (error) => {
+      const redisError = new RedisCacheError(error);
+      if (redisError.isSocketClosed()) {
+        this.logger.error(`Error occurred with Redis Connection when closing socket: ${redisError.message}`);
+      } else {
+        this.logger.error(`Error occurred with Redis Connection: ${redisError.fullError}`);
+      }
+    });
+  }
+
+  async connect(): Promise<void> {
+    await this.client.connect();
+    this.connected = true;
+  }
+
+  async disconnect(): Promise<void> {
+    await this.client.quit();
+    this.connected = false;
+  }
+
+  isConnected(): boolean {
+    return this.connected;
+  }
+
+  async getNumberOfConnections(): Promise<number> {
+    const list = await this.client.clientList();
+    return list.length;
+  }
+
+  getClient(): RedisClientType {
+    return this.client;
+  }
+}