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

@@ -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 { 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,6 +28,7 @@ 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'),
   };
 
   /**
@@ -37,12 +37,6 @@ export class RedisCache implements IRedisCacheClient {
    */
   private readonly logger: Logger;
 
-  /**
-   * The metrics register used for metrics tracking.
-   * @private
-   */
-  private readonly register: Registry;
-
   /**
    * The Redis client.
    * @private
@@ -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

@@ -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);
+  }
+}

packages/relay/src/lib/factories/cacheClientFactory.ts

@@ -19,6 +19,7 @@ import { IPAddressHbarSpendingPlanRepository } from './db/repositories/hbarLimit
 import { DebugImpl } from './debug';
 import { RpcMethodDispatcher } from './dispatcher';
 import { EthImpl } from './eth';
+import { CacheClientFactory } from './factories/cacheClientFactory';
 import { NetImpl } from './net';
 import { CacheService } from './services/cacheService/cacheService';
 import HAPIService from './services/hapiService/hapiService';
@@ -284,10 +285,13 @@ export class Relay {
 
     // Create CacheService with the connected Redis client (or undefined for LRU-only)
     this.cacheService = new CacheService(
-      this.logger.child({ name: 'cache-service' }),
+      CacheClientFactory.create(
+        this.logger.child({ name: 'cache-service' }),
+        this.register,
+        reservedKeys,
+        this.redisClient,
+      ),
       this.register,
-      reservedKeys,
-      this.redisClient,
     );
 
     // Create spending plan repositories