feat: integrates new lockService in sendRawTransaction (#4627)

Signed-off-by: Konstantina Blazhukova <[email protected]>
Signed-off-by: konstantinabl <[email protected]>
Co-authored-by: Logan Nguyen <[email protected]>
Signed-off-by: Konstantina Blazhukova <[email protected]>

Author: konstantinabl

package-lock.json

@@ -664,6 +664,11 @@ const _CONFIG = {
     required: false,
     defaultValue: 50,
   },
+  ENABLE_NONCE_ORDERING: {
+    type: 'boolean',
+    required: false,
+    defaultValue: false,
+  },
   USE_MIRROR_NODE_MODULARIZED_SERVICES: {
     type: 'boolean',
     required: false,

packages/config-service/src/services/globalConfig.ts

@@ -19,6 +19,7 @@ import {
   IBlockService,
   ICommonService,
   IContractService,
+  LockService,
   TransactionPoolService,
   TransactionService,
 } from './services';
@@ -126,6 +127,7 @@ export class EthImpl implements Eth {
     chain: string,
     public readonly cacheService: CacheService,
     storage: PendingTransactionStorage,
+    lockService: LockService,
   ) {
     this.chain = chain;
     this.logger = logger;
@@ -146,6 +148,7 @@ export class EthImpl implements Eth {
       logger,
       mirrorNodeClient,
       transactionPoolService,
+      lockService,
     );
     this.accountService = new AccountService(
       cacheService,

packages/relay/src/lib/eth.ts

@@ -20,6 +20,7 @@ import { DebugImpl } from './debug';
 import { RpcMethodDispatcher } from './dispatcher';
 import { EthImpl } from './eth';
 import { NetImpl } from './net';
+import { LockService, LockStrategyFactory } from './services';
 import { CacheService } from './services/cacheService/cacheService';
 import HAPIService from './services/hapiService/hapiService';
 import { HbarLimitService } from './services/hbarLimitService';
@@ -314,7 +315,7 @@ export class Relay {
       duration,
     );
 
-    // Create HAPI service
+    const lockService = new LockService(LockStrategyFactory.create(this.redisClient, this.logger));
     const hapiService = new HAPIService(this.logger, this.register, hbarLimitService);
     this.operatorAccountId = hapiService.getOperatorAccountId();
 
@@ -348,6 +349,7 @@ export class Relay {
       chainId,
       this.cacheService,
       storage,
+      lockService,
     );
 
     // Set up event listeners

packages/relay/src/lib/relay.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, TransactionPoolService } from '../../index';
 import { ITransactionService } from './ITransactionService';
 
 export class TransactionService implements ITransactionService {
@@ -46,6 +46,13 @@ export class TransactionService implements ITransactionService {
    */
   private readonly hapiService: HAPIService;
 
+  /**
+   * The lock service for managing transaction ordering.
+   * @private
+   * @readonly
+   */
+  private readonly lockService: LockService;
+
   /**
    * Logger instance for logging messages.
    * @private
@@ -86,6 +93,7 @@ export class TransactionService implements ITransactionService {
     logger: Logger,
     mirrorNodeClient: MirrorNodeClient,
     transactionPoolService: TransactionPoolService,
+    lockService: LockService,
   ) {
     this.cacheService = cacheService;
     this.chain = chain;
@@ -96,6 +104,7 @@ export class TransactionService implements ITransactionService {
     this.mirrorNodeClient = mirrorNodeClient;
     this.precheck = new Precheck(mirrorNodeClient, chain, transactionPoolService);
     this.transactionPoolService = transactionPoolService;
+    this.lockService = lockService;
   }
 
   /**
@@ -253,36 +262,60 @@ export class TransactionService implements ITransactionService {
 
     const transactionBuffer = Buffer.from(this.prune0x(transaction), 'hex');
     const parsedTx = Precheck.parseRawTransaction(transaction);
-    const networkGasPriceInWeiBars = Utils.addPercentageBufferToGasPrice(
-      await this.common.getGasPriceInWeibars(requestDetails),
-    );
+    let lockSessionKey: string | undefined;
+
+    // Acquire lock FIRST - before any side effects or async operations
+    // This ensures proper nonce ordering for transactions from the same sender
+    if (parsedTx.from) {
+      lockSessionKey = await this.lockService.acquireLock(parsedTx.from);
+    }
 
-    await this.validateRawTransaction(parsedTx, networkGasPriceInWeiBars, requestDetails);
+    try {
+      const networkGasPriceInWeiBars = Utils.addPercentageBufferToGasPrice(
+        await this.common.getGasPriceInWeibars(requestDetails),
+      );
 
-    // Save the transaction to the transaction pool before submitting it to the network
-    await this.transactionPoolService.saveTransaction(parsedTx.from!, parsedTx);
+      await this.validateRawTransaction(parsedTx, networkGasPriceInWeiBars, requestDetails);
 
-    /**
-     * Note: If the USE_ASYNC_TX_PROCESSING feature flag is enabled,
-     * the transaction hash is calculated and returned immediately after passing all prechecks.
-     * All transaction processing logic is then handled asynchronously in the background.
-     */
-    const useAsyncTxProcessing = ConfigService.get('USE_ASYNC_TX_PROCESSING');
-    if (useAsyncTxProcessing) {
-      this.sendRawTransactionProcessor(transactionBuffer, parsedTx, networkGasPriceInWeiBars, requestDetails);
-      return Utils.computeTransactionHash(transactionBuffer);
-    }
+      // Save the transaction to the transaction pool before submitting it to the network
+      await this.transactionPoolService.saveTransaction(parsedTx.from!, parsedTx);
 
-    /**
-     * Note: If the USE_ASYNC_TX_PROCESSING feature flag is disabled,
-     * wait for all transaction processing logic to complete before returning the transaction hash.
-     */
-    return await this.sendRawTransactionProcessor(
-      transactionBuffer,
-      parsedTx,
-      networkGasPriceInWeiBars,
-      requestDetails,
-    );
+      /**
+       * Note: If the USE_ASYNC_TX_PROCESSING feature flag is enabled,
+       * the transaction hash is calculated and returned immediately after passing all prechecks.
+       * All transaction processing logic is then handled asynchronously in the background.
+       */
+      const useAsyncTxProcessing = ConfigService.get('USE_ASYNC_TX_PROCESSING');
+      if (useAsyncTxProcessing) {
+        // Fire and forget - lock will be released after consensus submission
+        this.sendRawTransactionProcessor(
+          transactionBuffer,
+          parsedTx,
+          networkGasPriceInWeiBars,
+          lockSessionKey,
+          requestDetails,
+        );
+        return Utils.computeTransactionHash(transactionBuffer);
+      }
+
+      /**
+       * Note: If the USE_ASYNC_TX_PROCESSING feature flag is disabled,
+       * wait for all transaction processing logic to complete before returning the transaction hash.
+       */
+      return await this.sendRawTransactionProcessor(
+        transactionBuffer,
+        parsedTx,
+        networkGasPriceInWeiBars,
+        lockSessionKey,
+        requestDetails,
+      );
+    } catch (error) {
+      // Release lock on any error during validation or prechecks
+      if (lockSessionKey) {
+        await this.lockService.releaseLock(parsedTx.from!, lockSessionKey);
+      }
+      throw error;
+    }
   }
 
   /**
@@ -472,12 +505,14 @@ 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 | null} lockSessionKey - The session key for the acquired lock, null if no lock was acquired.
    * @returns {Promise<string | JsonRpcError>} A promise that resolves to the transaction hash if successful, or a JsonRpcError if an error occurs.
    */
   async sendRawTransactionProcessor(
     transactionBuffer: Buffer,
     parsedTx: EthersTransaction,
     networkGasPriceInWeiBars: number,
+    lockSessionKey: string | undefined,
     requestDetails: RequestDetails,
   ): Promise<string | JsonRpcError> {
     let sendRawTransactionError: any;
@@ -495,6 +530,9 @@ export class TransactionService implements ITransactionService {
       requestDetails,
     );
 
+    if (lockSessionKey) {
+      await this.lockService.releaseLock(originalCallerAddress, lockSessionKey);
+    }
     // Remove the transaction from the transaction pool after submission
     await this.transactionPoolService.removeTransaction(originalCallerAddress, parsedTx.serialized);
 

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

@@ -6,6 +6,7 @@ import { randomUUID } from 'crypto';
 import { LRUCache } from 'lru-cache';
 import { Logger } from 'pino';
 
+import { LockStrategy } from '../../types/lock';
 import { LockService } from './LockService';
 
 /**
@@ -24,7 +25,7 @@ export interface LockState {
  * 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 {
+export class LocalLockStrategy implements LockStrategy {
   /**
    * LRU cache of lock states, keyed by address.
    */
@@ -55,12 +56,11 @@ export class LocalLockStrategy {
    * @param address - The key representing the resource to lock
    * @returns A session key identifying the current lock owner
    */
-  async acquireLock(address: string): Promise<string> {
+  async acquireLock(address: string): Promise<string | undefined> {
+    const sessionKey = randomUUID();
     if (this.logger.isLevelEnabled('debug')) {
-      this.logger.debug(`Acquiring lock for address ${address}.`);
+      this.logger.debug(`Acquiring lock for address ${address} and sessionkey ${sessionKey}.`);
     }
-
-    const sessionKey = randomUUID();
     const state = this.getOrCreateState(address);
 
     // Acquire the mutex (this will block until available)
@@ -85,16 +85,19 @@ export class LocalLockStrategy {
    * @param sessionKey - The session key of the lock holder
    */
   async releaseLock(address: string, sessionKey: string): Promise<void> {
-    const state = this.localLockStates.get(address);
-
-    if (this.logger.isLevelEnabled('debug') && state?.acquiredAt) {
-      const holdTime = Date.now() - state.acquiredAt;
-      this.logger.debug(`Releasing lock for address ${address} and session key ${sessionKey} held for ${holdTime}ms.`);
-    }
-
-    // Ensure only the lock owner can release
-    if (state?.sessionKey === sessionKey) {
-      await this.doRelease(state);
+    const normalizedAddress = LockService.normalizeAddress(address);
+    const state = this.localLockStates.get(normalizedAddress);
+    if (state) {
+      // Ensure only the lock owner can release
+      if (state.sessionKey === sessionKey) {
+        await this.doRelease(state);
+        if (this.logger.isLevelEnabled('debug')) {
+          const holdTime = Date.now() - state.acquiredAt!;
+          this.logger.debug(
+            `Releasing lock for address ${address} and session key ${sessionKey} held for ${holdTime}ms.`,
+          );
+        }
+      }
     }
   }
 
@@ -144,7 +147,8 @@ export class LocalLockStrategy {
    * @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);
+    const normalizedAddress = LockService.normalizeAddress(address);
+    const state = this.localLockStates.get(normalizedAddress);
 
     if (state?.sessionKey === sessionKey) {
       await this.doRelease(state);

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

@@ -1,5 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
+import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
+
 import { LockStrategy } from '../../types';
 
 /**
@@ -28,7 +30,10 @@ export class LockService {
    * @param address - The sender address to acquire the lock for.
    * @returns A promise that resolves to a unique session key, or null if acquisition fails (fail open).
    */
-  async acquireLock(address: string): Promise<string | null> {
+  async acquireLock(address: string): Promise<string | undefined> {
+    if (!ConfigService.get('ENABLE_NONCE_ORDERING')) {
+      return;
+    }
     return await this.strategy.acquireLock(address);
   }
 
@@ -40,7 +45,9 @@ export class LockService {
    * @param sessionKey - The session key obtained during lock acquisition.
    */
   async releaseLock(address: string, sessionKey: string): Promise<void> {
-    await this.strategy.releaseLock(address, sessionKey);
+    if (ConfigService.get('ENABLE_NONCE_ORDERING')) {
+      await this.strategy.releaseLock(address, sessionKey);
+    }
   }
 
   /**

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

@@ -39,7 +39,7 @@ export class RedisLockStrategy implements LockStrategy {
    * @param address - The sender address to acquire the lock for (will be normalized).
    * @returns A promise that resolves to a unique session key upon successful acquisition, or null if acquisition fails (fail open).
    */
-  async acquireLock(address: string): Promise<string | null> {
+  async acquireLock(address: string): Promise<string | undefined> {
     const sessionKey = this.generateSessionKey();
     const lockKey = this.getLockKey(address);
     const queueKey = this.getQueueKey(address);
@@ -86,7 +86,7 @@ export class RedisLockStrategy implements LockStrategy {
       }
     } catch (error) {
       this.logger.error(error, `Failed to acquire lock: address=${address}, sessionKey=${sessionKey}. Failing open.`);
-      return null;
+      return;
     } finally {
       // Always remove from queue if we joined it (whether success or failure)
       if (joinedQueue) {

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

@@ -15,7 +15,7 @@ export interface LockStrategy {
    * @param address - The address to acquire the lock for (will be normalized by implementation).
    * @returns A promise that resolves to a unique session key upon successful acquisition, or null if acquisition fails (fail open).
    */
-  acquireLock(address: string): Promise<string | null>;
+  acquireLock(address: string): Promise<string | undefined>;
 
   /**
    * Releases a lock for the specified address.