feat: create LocalLockStrategy using LRU and async-mutex

packages/relay/src/lib/services/ethService/transactionService/TransactionService.ts

@@ -21,7 +21,7 @@ import { Precheck } from '../../../precheck';
 import { ITransactionReceipt, RequestDetails, TypedEvents } from '../../../types';
 import { CacheService } from '../../cacheService/cacheService';
 import HAPIService from '../../hapiService/hapiService';
-import { ICommonService, TransactionPoolService } from '../../index';
+import { ICommonService, LockService, LockStrategyFactory, TransactionPoolService } from '../../index';
 import { ITransactionService } from './ITransactionService';
 
 export class TransactionService implements ITransactionService {
@@ -66,8 +66,18 @@ export class TransactionService implements ITransactionService {
    */
   private readonly precheck: Precheck;
 
+  /**
+   * Service responsible for managing pending transactions.
+   */
   private readonly transactionPoolService: TransactionPoolService;
 
+  /**
+   * Service that provides mechanisms for acquiring and releasing locks
+   * to ensure thread-safe and concurrent operation handling across
+   * asynchronous processes.
+   */
+  private readonly lockService: LockService;
+
   /**
    * The ID of the chain, as a hex string, as it would be returned in a JSON-RPC call.
    * @private
@@ -96,6 +106,7 @@ export class TransactionService implements ITransactionService {
     this.mirrorNodeClient = mirrorNodeClient;
     this.precheck = new Precheck(mirrorNodeClient, chain, transactionPoolService);
     this.transactionPoolService = transactionPoolService;
+    this.lockService = new LockService(LockStrategyFactory.create(undefined, logger), logger);
   }
 
   /**
@@ -472,6 +483,7 @@ export class TransactionService implements ITransactionService {
    * @param {EthersTransaction} parsedTx - The parsed Ethereum transaction object.
    * @param {number} networkGasPriceInWeiBars - The current network gas price in wei bars.
    * @param {RequestDetails} requestDetails - Details of the request for logging and tracking purposes.
+   * @param {string} sessionKey - The key that is used as identifier in the lock service
    * @returns {Promise<string | JsonRpcError>} A promise that resolves to the transaction hash if successful, or a JsonRpcError if an error occurs.
    */
   async sendRawTransactionProcessor(
@@ -484,6 +496,8 @@ export class TransactionService implements ITransactionService {
 
     const originalCallerAddress = parsedTx.from?.toString() || '';
 
+    const sessionKey = await this.lockService.acquireLock(parsedTx.from!);
+
     this.eventEmitter.emit('eth_execution', {
       method: constants.ETH_SEND_RAW_TRANSACTION,
     });
@@ -497,6 +511,7 @@ export class TransactionService implements ITransactionService {
 
     // Remove the transaction from the transaction pool after submission
     await this.transactionPoolService.removeTransaction(originalCallerAddress, parsedTx.serialized);
+    await this.lockService.releaseLock(parsedTx.from!, sessionKey);
 
     sendRawTransactionError = error;
 

packages/relay/src/lib/services/lockService/LocalLockStrategy.ts

@@ -0,0 +1,168 @@
+// SPDX-License-Identifier: Apache-2.0
+
+import { Mutex } from 'async-mutex';
+import { randomUUID } from 'crypto';
+import { LRUCache } from 'lru-cache';
+import { Logger } from 'pino';
+
+/**
+ * Represents the internal state for a lock associated with a given address.
+ */
+interface LockState {
+  mutex: Mutex;
+  sessionKey: string | null;
+  acquiredAt: number | null;
+  maxLockTime: NodeJS.Timeout | null;
+}
+
+/**
+ * Implements a local, in-memory locking strategy.
+ *
+ * Each unique "address" gets its own mutex to ensure only one session can hold
+ * the lock at a time. Locks are auto-expiring and stored in an LRU cache.
+ */
+export class LocalLockStrategy {
+  /**
+   * Maximum number of lock entries stored in memory.
+   * Prevents unbounded memory growth.
+   */
+  public static LOCAL_LOCK_MAX_ENTRIES: number = 1_000; // Max 1000 addresses
+
+  /**
+   * Time-to-live for each lock entry in the cache (in milliseconds).
+   */
+  public static LOCAL_LOCK_TTL: number = 300_000; // 5 minutes
+
+  /**
+   * Seconds for auto-release if lock not manually released
+   */
+  public static LOCAL_LOCK_MAX_LOCK_TIME: number = 30_000; // 30 secs
+
+  /**
+   * LRU cache of lock states, keyed by address.
+   */
+  private localLockStates = new LRUCache<string, LockState>({
+    max: LocalLockStrategy.LOCAL_LOCK_MAX_ENTRIES,
+    ttl: LocalLockStrategy.LOCAL_LOCK_TTL,
+  });
+
+  /**
+   * Logger.
+   *
+   * @private
+   */
+  private readonly logger: Logger;
+
+  /**
+   * Creates a new LocalLockStrategy instance.
+   *
+   * @param logger - The logger
+   */
+  constructor(logger: Logger) {
+    this.logger = logger;
+  }
+
+  /**
+   * Acquire a lock for a specific address.
+   * Waits until the lock is available (blocking if another session holds it).
+   *
+   * @param address - The key representing the resource to lock
+   * @returns A session key identifying the current lock owner
+   */
+  async acquireLock(address: string): Promise<string> {
+    const sessionKey = randomUUID();
+    const state = this.getOrCreateState(address);
+
+    // Acquire the mutex (this will block until available)
+    await state.mutex.acquire();
+
+    // Record lock ownership metadata
+    state.sessionKey = sessionKey;
+    state.acquiredAt = Date.now();
+
+    // Start a 30-second timer to auto-release if lock not manually released
+    state.maxLockTime = setTimeout(() => {
+      this.forceReleaseExpiredLock(address, sessionKey);
+    }, LocalLockStrategy.LOCAL_LOCK_MAX_LOCK_TIME);
+
+    return sessionKey;
+  }
+
+  /**
+   * Release a previously acquired lock, if the session key matches the current owner.
+   *
+   * @param address - The locked resource key
+   * @param sessionKey - The session key of the lock holder
+   */
+  async releaseLock(address: string, sessionKey: string): Promise<void> {
+    const state = this.localLockStates.get(address);
+
+    // Ensure only the lock owner can release
+    if (state?.sessionKey !== sessionKey) {
+      return; // Not the owner — safely ignore
+    }
+
+    // Perform cleanup and release
+    await this.doRelease(state);
+  }
+
+  /**
+   * Retrieve an existing lock state for the given address, or create a new one if it doesn't exist.
+   *
+   * @param address - Unique identifier for the lock
+   * @returns The LockState object associated with the address
+   */
+  private getOrCreateState(address: string): LockState {
+    if (!this.localLockStates.has(address)) {
+      this.localLockStates.set(address, {
+        mutex: new Mutex(),
+        sessionKey: null,
+        acquiredAt: null,
+        maxLockTime: null,
+      });
+    }
+
+    return this.localLockStates.get(address)!;
+  }
+
+  /**
+   * Internal helper to perform cleanup and release the mutex.
+   *
+   * @param state - The LockState instance to reset and release
+   */
+  private async doRelease(state: LockState): Promise<void> {
+    // Clear timeout first
+    clearTimeout(state.maxLockTime!);
+
+    // Reset state
+    state.sessionKey = null;
+    state.maxLockTime = null;
+    state.acquiredAt = null;
+
+    // Release the mutex lock
+    state.mutex.release();
+  }
+
+  /**
+   * Forcefully release a lock that has exceeded its maximum execution time.
+   * Used by the timeout set during `acquireLock`.
+   *
+   * @param address - The resource key associated with the lock
+   * @param sessionKey - The session key to verify ownership before releasing
+   */
+  private async forceReleaseExpiredLock(address: string, sessionKey: string): Promise<void> {
+    const state = this.localLockStates.get(address);
+
+    // Ensure the session still owns the lock before force-releasing
+    if (!state || state.sessionKey !== sessionKey) {
+      return; // Already released or lock reassigned
+    }
+
+    if (this.logger.isLevelEnabled('debug')) {
+      const holdTime = Date.now() - state.acquiredAt!;
+      this.logger.debug(`Force releasing expired local lock for address ${address} held for ${holdTime}ms.`);
+    }
+
+    await this.doRelease(state);
+  }
+}

packages/relay/src/lib/services/lockService/LockService.ts

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
-import { LockStrategy } from '../../types/lock';
+import { Logger } from 'pino';
+
+import { LockStrategy } from '../../types';
 
 /**
  * Service that manages transaction ordering through distributed locking.
@@ -12,13 +14,22 @@ export class LockService {
    */
   private readonly strategy: LockStrategy;
 
+  /**
+   * Logger
+   *
+   * @private
+   */
+  private readonly logger: Logger;
+
   /**
    * Creates a new LockService instance.
    *
    * @param strategy - The lock strategy implementation to use.
+   * @param logger - The logger
    */
-  constructor(strategy: LockStrategy) {
+  constructor(strategy: LockStrategy, logger: Logger) {
     this.strategy = strategy;
+    this.logger = logger;
   }
 
   /**
@@ -29,6 +40,10 @@ export class LockService {
    * @returns A promise that resolves to a unique session key.
    */
   async acquireLock(address: string): Promise<string> {
+    if (this.logger.isLevelEnabled('debug')) {
+      this.logger.debug(`Acquiring lock for address ${address}.`);
+    }
+
     return await this.strategy.acquireLock(address);
   }
 
@@ -40,6 +55,10 @@ export class LockService {
    * @param sessionKey - The session key obtained during lock acquisition.
    */
   async releaseLock(address: string, sessionKey: string): Promise<void> {
+    if (this.logger.isLevelEnabled('debug')) {
+      this.logger.debug(`Releasing lock for address ${address} and session key ${sessionKey}.`);
+    }
+
     await this.strategy.releaseLock(address, sessionKey);
   }
 }

packages/relay/src/lib/services/lockService/LockStrategyFactory.ts

@@ -3,7 +3,8 @@
 import { Logger } from 'pino';
 import { RedisClientType } from 'redis';
 
-import { LockStrategy } from '../../types/lock';
+import { LockStrategy } from '../../types';
+import { LocalLockStrategy } from './LocalLockStrategy';
 
 /**
  * Factory for creating LockStrategy instances.
@@ -20,13 +21,13 @@ export class LockStrategyFactory {
    * @param logger - Logger instance for the lock strategy.
    * @returns A LockStrategy implementation.
    */
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+
   static create(redisClient: RedisClientType | undefined, logger: Logger): LockStrategy {
     // TODO: Remove placeholder errors once strategies are implemented
     if (redisClient) {
-      throw new Error('Redis lock strategy not yet implemented');
+      // throw new Error('Redis lock strategy not yet implemented');
     }
 
-    throw new Error('Local lock strategy not yet implemented');
+    return new LocalLockStrategy(logger);
   }
 }