feat: add cache factory & drop redis fallback & unify client int (#4558)

Signed-off-by: Mariusz Jasuwienas <[email protected]>

Author: jasuwienas

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

@@ -4,8 +4,11 @@ export interface ICacheClient {
   keys(pattern: string, callingMethod: string): Promise<string[]>;
   get(key: string, callingMethod: string): Promise<any>;
   set(key: string, value: any, callingMethod: string, ttl?: number): Promise<void>;
-  multiSet(keyValuePairs: Record<string, any>, callingMethod: string): Promise<void>;
+  multiSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number | undefined): Promise<void>;
   pipelineSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number | undefined): Promise<void>;
   delete(key: string, callingMethod: string): Promise<void>;
   clear(): Promise<void>;
+  incrBy(key: string, amount: number, callingMethod: string): Promise<number>;
+  rPush(key: string, value: any, callingMethod: string): Promise<number>;
+  lRange<T = any>(key: string, start: number, end: number, callingMethod: string): Promise<T[]>;
 }

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

@@ -72,6 +72,7 @@ export class LocalLRUCache implements ICacheClient {
    * @constructor
    * @param {Logger} logger - The logger instance to be used for logging.
    * @param {Registry} register - The registry instance used for metrics tracking.
+   * @param {Set<string>} reservedKeys - These are the cache keys delegated to the reserved cache.
    */
   public constructor(logger: Logger, register: Registry, reservedKeys: Set<string> = new Set()) {
     this.cache = new LRUCache(this.options);
@@ -137,8 +138,9 @@ export class LocalLRUCache implements ICacheClient {
    * @param key - The key to check the remaining TTL for.
    * @param callingMethod - The name of the method calling the cache.
    * @returns The remaining TTL in milliseconds.
+   * @private
    */
-  public async getRemainingTtl(key: string, callingMethod: string): Promise<number> {
+  private async getRemainingTtl(key: string, callingMethod: string): Promise<number> {
     const prefixedKey = this.prefixKey(key);
     const cache = this.getCacheInstance(key);
     const remainingTtl = cache.getRemainingTTL(prefixedKey); // in milliseconds
@@ -280,6 +282,70 @@ export class LocalLRUCache implements ICacheClient {
     return matchingKeys.map((key) => key.substring(LocalLRUCache.CACHE_KEY_PREFIX.length));
   }
 
+  /**
+   * Increments a value in the cache.
+   *
+   * @param key The key to increment
+   * @param amount The amount to increment by
+   * @param callingMethod The name of the calling method
+   * @returns The value of the key after incrementing
+   */
+  public async incrBy(key: string, amount: number, callingMethod: string): Promise<number> {
+    const value = await this.get(key, callingMethod);
+    const newValue = value + amount;
+    const remainingTtl = await this.getRemainingTtl(key, callingMethod);
+    await this.set(key, newValue, callingMethod, remainingTtl);
+    return newValue;
+  }
+
+  /**
+   * Retrieves a range of elements from a list in the cache.
+   *
+   * @param key The key of the list
+   * @param start The start index
+   * @param end The end index
+   * @param callingMethod The name of the calling method
+   * @returns The list of elements in the range
+   */
+  public async lRange(key: string, start: number, end: number, callingMethod: string): Promise<any[]> {
+    const values = (await this.get(key, callingMethod)) ?? [];
+    if (!Array.isArray(values)) {
+      throw new Error(`Value at key ${key} is not an array`);
+    }
+    if (end < 0) {
+      end = values.length + end;
+    }
+    return values.slice(start, end + 1);
+  }
+
+  /**
+   * Pushes a value to the end of a list in the cache.
+   *
+   * @param key The key of the list
+   * @param value The value to push
+   * @param callingMethod The name of the calling method
+   * @returns The length of the list after pushing
+   */
+  public async rPush(key: string, value: any, callingMethod: string): Promise<number> {
+    const values = (await this.get(key, callingMethod)) ?? [];
+    if (!Array.isArray(values)) {
+      throw new Error(`Value at key ${key} is not an array`);
+    }
+    values.push(value);
+    const remainingTtl = await this.getRemainingTtl(key, callingMethod);
+    await this.set(key, values, callingMethod, remainingTtl);
+    return values.length;
+  }
+
+  /**
+   * Returns the appropriate cache instance for the given key.
+   * If a reserved cache exists and the key is marked as reserved,
+   * the reserved cache is returned; otherwise the default cache is used.
+   *
+   * @param key -  The cache key being accessed.
+   * @returns The selected cache instance
+   * @private
+   */
   private getCacheInstance(key: string): LRUCache<string, any> {
     return this.reservedCache && this.reservedKeys.has(key) ? this.reservedCache : this.cache;
   }

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

@@ -0,0 +1,3 @@
+// SPDX-License-Identifier: Apache-2.0
+
+export { SafeRedisCache as RedisCache } from './safeRedisCache';

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

@@ -2,16 +2,15 @@
 
 import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
 import { Logger } from 'pino';
-import { Registry } from 'prom-client';
 import { RedisClientType } from 'redis';
 
-import { Utils } from '../../../utils';
-import { IRedisCacheClient } from './IRedisCacheClient';
+import { Utils } from '../../../../utils';
+import { ICacheClient } from '../ICacheClient';
 
 /**
  * A class that provides caching functionality using Redis.
  */
-export class RedisCache implements IRedisCacheClient {
+export class RedisCache implements ICacheClient {
   /**
    * Prefix used to namespace all keys managed by this cache.
    *
@@ -29,19 +28,14 @@ export class RedisCache implements IRedisCacheClient {
   private readonly options = {
     // Max time to live in ms, for items before they are considered stale.
     ttl: ConfigService.get('CACHE_TTL'),
+    multiSetEnabled: ConfigService.get('MULTI_SET'),
   };
 
   /**
    * The logger used for logging all output from this class.
    * @private
    */
-  private readonly logger: Logger;
-
-  /**
-   * The metrics register used for metrics tracking.
-   * @private
-   */
-  private readonly register: Registry;
+  protected readonly logger: Logger;
 
   /**
    * The Redis client.
@@ -53,11 +47,10 @@ export class RedisCache implements IRedisCacheClient {
    * Creates an instance of `RedisCache`.
    *
    * @param {Logger} logger - The logger instance.
-   * @param {Registry} register - The metrics registry.
+   * @param {RedisClientType} client
    */
-  public constructor(logger: Logger, register: Registry, client: RedisClientType) {
+  public constructor(logger: Logger, client: RedisClientType) {
     this.logger = logger;
-    this.register = register;
     this.client = client;
   }
 
@@ -129,9 +122,11 @@ export class RedisCache implements IRedisCacheClient {
    *
    * @param keyValuePairs - An object where each property is a key and its value is the value to be cached.
    * @param callingMethod - The name of the calling method.
+   * @param [ttl] - The time-to-live (expiration) of the cache item in milliseconds. Used in fallback to pipelineSet.
    * @returns A Promise that resolves when the values are cached.
    */
-  async multiSet(keyValuePairs: Record<string, any>, callingMethod: string): Promise<void> {
+  async multiSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number): Promise<void> {
+    if (!this.options.multiSetEnabled) return this.pipelineSet(keyValuePairs, callingMethod, ttl);
     // Serialize values and add prefix
     const serializedKeyValuePairs: Record<string, string> = {};
     for (const [key, value] of Object.entries(keyValuePairs)) {

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

@@ -0,0 +1,172 @@
+// SPDX-License-Identifier: Apache-2.0
+
+import { RedisCacheError } from '../../../errors/RedisCacheError';
+import { RedisCache } from './redisCache';
+
+/**
+ * A safer wrapper around {@link RedisCache} which is responsible for:
+ *  - ignoring all Redis command errors.
+ *  - logging all errors,
+ *  - returning default values in cases of failures.
+ *
+ * Thanks to that our application will be able to continue functioning even with Redis being down...
+ */
+export class SafeRedisCache extends RedisCache {
+  /**
+   * Retrieves a value from the cache.
+   *
+   * This method wraps {@link RedisCache.get} and ensures `null` is returned instead of throwing error.
+   *
+   * @param key - The cache key.
+   * @param callingMethod - Name of the method making the request (for logging).
+   * @returns The cached value, or `null` if Redis fails or the value does not exist.
+   */
+  async get(key: string, callingMethod: string): Promise<any> {
+    return await this.safeCall(() => super.get(key, callingMethod), null);
+  }
+
+  /**
+   /**
+   * Stores a value in the cache safely.
+   *
+   * Wraps {@link RedisCache.set} and suppresses Redis errors.
+   * On failure, nothing is thrown and the error is logged.
+   *
+   * @param key - The cache key.
+   * @param value - The value to store.
+   * @param callingMethod - Name of the calling method.
+   * @param ttl - Optional TTL in milliseconds.
+   */
+  async set(key: string, value: any, callingMethod: string, ttl?: number): Promise<void> {
+    await this.safeCall(() => super.set(key, value, callingMethod, ttl), undefined);
+  }
+
+  /**
+   * Stores multiple key-value pairs safely.
+   *
+   * Wraps {@link RedisCache.multiSet} with error suppression.
+   *
+   * @param keyValuePairs - Object of key-value pairs to set.
+   * @param callingMethod - Name of the calling method.
+   * @param ttl - Optional TTL used in fallback pipeline mode.
+   */
+  async multiSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number): Promise<void> {
+    await this.safeCall(() => super.multiSet(keyValuePairs, callingMethod, ttl), undefined);
+  }
+
+  /**
+   * Performs a pipelined multi-set operation safely.
+   *
+   * Wraps {@link RedisCache.pipelineSet} with error suppression.
+   *
+   * @param keyValuePairs - Key-value pairs to write.
+   * @param callingMethod - Name of the calling method.
+   * @param ttl - Optional TTL.
+   */
+  async pipelineSet(keyValuePairs: Record<string, any>, callingMethod: string, ttl?: number): Promise<void> {
+    await this.safeCall(() => super.pipelineSet(keyValuePairs, callingMethod, ttl), undefined);
+  }
+
+  /**
+   * Deletes a value from the cache safely.
+   *
+   * Wraps {@link RedisCache.delete} with error suppression.
+   *
+   * @param key - Key to delete.
+   * @param callingMethod - Name of the calling method.
+   */
+  async delete(key: string, callingMethod: string): Promise<void> {
+    await this.safeCall(() => super.delete(key, callingMethod), undefined);
+  }
+
+  /**
+   * Increments a numeric value safely.
+   *
+   * Wraps {@link RedisCache.incrBy}.
+   * On failure, returns the `amount` argument as fallback.
+   *
+   * @param key - Key to increment.
+   * @param amount - Increment amount.
+   * @param callingMethod - Name of the calling method.
+   * @returns The incremented value or the fallback (amount) if Redis fails.
+   */
+  async incrBy(key: string, amount: number, callingMethod: string): Promise<number> {
+    return await this.safeCall(() => super.incrBy(key, amount, callingMethod), amount);
+  }
+
+  /**
+   * Retrieves a list slice safely.
+   *
+   * Wraps {@link RedisCache.lRange}.
+   * On error, returns an empty array.
+   *
+   * @param key - List key.
+   * @param start - Start index.
+   * @param end - End index.
+   * @param callingMethod - Name of the calling method.
+   * @returns List of elements, or an empty array on failure.
+   */
+  async lRange(key: string, start: number, end: number, callingMethod: string): Promise<any[]> {
+    return await this.safeCall(() => super.lRange(key, start, end, callingMethod), []);
+  }
+
+  /**
+   * Pushes a value to a list safely.
+   *
+   * Wraps {@link RedisCache.rPush}.
+   * Returns `0` on failure.
+   *
+   * @param key - List key.
+   * @param value - Value to push.
+   * @param callingMethod - Name of the calling method.
+   * @returns The new list length, or `0` if Redis fails.
+   */
+  async rPush(key: string, value: any, callingMethod: string): Promise<number> {
+    return await this.safeCall(() => super.rPush(key, value, callingMethod), 0);
+  }
+
+  /**
+   * Retrieves keys matching a pattern safely.
+   *
+   * Wraps {@link RedisCache.keys}.
+   * Returns an empty array on error.
+   *
+   * @param pattern - Match pattern.
+   * @param callingMethod - Name of the calling method.
+   * @returns Array of matched keys (prefix removed), or empty array on error.
+   */
+  async keys(pattern: string, callingMethod: string): Promise<string[]> {
+    return await this.safeCall(() => super.keys(pattern, callingMethod), []);
+  }
+
+  /**
+   * Clears all cache keys safely.
+   *
+   * Wraps {@link RedisCache.clear}.
+   * Any Redis failure is logged and ignored.
+   */
+
+  async clear(): Promise<void> {
+    await this.safeCall(() => super.clear(), null);
+  }
+
+  /**
+   * Executes a Redis call safely.
+   *
+   * This is the core safety mechanism of {@link SafeRedisCache}.
+   *
+   * @template T The expected return type.
+   * @param fn - Function containing the Redis call.
+   * @param fallback - Value to return if an error occurs.
+   * @returns The result of `fn()` or the fallback.
+   */
+  async safeCall<T>(fn: () => Promise<T>, fallback: T): Promise<T> {
+    try {
+      return await fn();
+    } catch (error) {
+      const redisError = new RedisCacheError(error);
+      this.logger.error(redisError, 'Error occurred while getting the cache from Redis.');
+      return fallback;
+    }
+  }
+}

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

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 export * from './cache/localLRUCache';
-export * from './cache/redisCache';
+export * from './cache/redisCache/index';
 export * from './mirrorNodeClient';
 export * from './sdkClient';

packages/relay/src/lib/clients/index.ts

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: Apache-2.0
+
+import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
+import type { Logger } from 'pino';
+import { Registry } from 'prom-client';
+import { RedisClientType } from 'redis';
+
+import { LocalLRUCache, RedisCache } from '../clients';
+import { ICacheClient } from '../clients/cache/ICacheClient';
+
+export class CacheClientFactory {
+  static create(
+    logger: Logger,
+    register: Registry = new Registry(),
+    reservedKeys: Set<string> = new Set(),
+    redisClient?: RedisClientType,
+  ): ICacheClient {
+    return !ConfigService.get('TEST') && redisClient !== undefined
+      ? new RedisCache(logger.child({ name: 'redisCache' }), redisClient)
+      : new LocalLRUCache(logger.child({ name: 'localLRUCache' }), register, reservedKeys);
+  }
+}