runExclusive instead of manual

Signed-off-by: Quiet Node <[email protected]>

Author: quiet-node

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

@@ -27,7 +27,7 @@ import { Logger } from 'pino';
 
 import { weibarHexToTinyBarInt } from '../../formatters';
 import { Utils } from '../../utils';
-import { CommonService, RawTxSynchronizeService } from '../services';
+import { CommonService } from '../services';
 import { HbarLimitService } from '../services/hbarLimitService';
 import { ITransactionRecordMetric, RequestDetails, TypedEvents } from '../types';
 import constants from './../constants';
@@ -145,7 +145,6 @@ export class SDKClient {
     originalCallerAddress: string,
     networkGasPriceInWeiBars: number,
     currentNetworkExchangeRateInCents: number,
-    rawTxSynchronizeService: RawTxSynchronizeService,
   ): Promise<{ txResponse: TransactionResponse; fileId: FileId | null }> {
     const jumboTxEnabled = ConfigService.get('JUMBO_TX_ENABLED');
     const ethereumTransactionData: EthereumTransactionData = EthereumTransactionData.fromBytes(transactionBuffer);
@@ -191,7 +190,6 @@ export class SDKClient {
         requestDetails,
         true,
         originalCallerAddress,
-        rawTxSynchronizeService,
       ),
     };
   }
@@ -278,7 +276,6 @@ export class SDKClient {
     requestDetails: RequestDetails,
     shouldThrowHbarLimit: boolean,
     originalCallerAddress: string,
-    rawTxSynchronizeService?: RawTxSynchronizeService,
     estimatedTxFee?: number,
   ): Promise<TransactionResponse> {
     const txConstructorName = transaction.constructor.name;
@@ -336,11 +333,6 @@ export class SDKClient {
       }
       return transactionResponse;
     } finally {
-      // Eventually release the transaction mutex lock if it was acquired by the sender
-      if (rawTxSynchronizeService) {
-        await rawTxSynchronizeService.releaseLock(originalCallerAddress);
-      }
-
       if (transactionId?.length) {
         this.eventEmitter.emit('execute_transaction', {
           transactionId,

packages/relay/src/lib/eth.ts

@@ -151,7 +151,6 @@ export class EthImpl implements Eth {
       hapiService,
       logger,
       mirrorNodeClient,
-      this.rawTxSynchronizeService,
     );
   }
 
@@ -960,7 +959,9 @@ export class EthImpl implements Eth {
     0: { type: 'hex', required: true },
   })
   async sendRawTransaction(transaction: string, requestDetails: RequestDetails): Promise<string | JsonRpcError> {
-    return await this.transactionService.sendRawTransaction(transaction, requestDetails);
+    return await this.rawTxSynchronizeService.runExclusive(transaction, async () => {
+      return await this.transactionService.sendRawTransaction(transaction, requestDetails);
+    });
   }
 
   /**

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, RawTxSynchronizeService } from '../../index';
+import { ICommonService } from '../../index';
 import { ITransactionService } from './ITransactionService';
 
 export class TransactionService implements ITransactionService {
@@ -72,13 +72,6 @@ export class TransactionService implements ITransactionService {
    */
   private readonly chain: string;
 
-  /**
-   * The service responsible for synchronizing raw transaction submissions to ensure nonce ordering.
-   * @private
-   * @readonly
-   */
-  private readonly rawTxSynchronizeService: RawTxSynchronizeService;
-
   /**
    * Constructor for the TransactionService class.
    */
@@ -90,7 +83,6 @@ export class TransactionService implements ITransactionService {
     hapiService: HAPIService,
     logger: Logger,
     mirrorNodeClient: MirrorNodeClient,
-    rawTxSynchronizeService: RawTxSynchronizeService,
   ) {
     this.cacheService = cacheService;
     this.chain = chain;
@@ -99,7 +91,6 @@ export class TransactionService implements ITransactionService {
     this.hapiService = hapiService;
     this.logger = logger;
     this.mirrorNodeClient = mirrorNodeClient;
-    this.rawTxSynchronizeService = rawTxSynchronizeService;
     this.precheck = new Precheck(mirrorNodeClient, logger, chain);
   }
 
@@ -256,12 +247,6 @@ export class TransactionService implements ITransactionService {
 
     const transactionBuffer = Buffer.from(this.prune0x(transaction), 'hex');
     const parsedTx = Precheck.parseRawTransaction(transaction);
-
-    // Acquire a lock for the sender before any side effects or asynchronous calls to ensure proper nonce ordering
-    if (parsedTx.from) {
-      await this.rawTxSynchronizeService.acquireLock(parsedTx.from);
-    }
-
     const networkGasPriceInWeiBars = Utils.addPercentageBufferToGasPrice(
       await this.common.getGasPriceInWeibars(requestDetails),
     );
@@ -275,13 +260,7 @@ export class TransactionService implements ITransactionService {
      */
     const useAsyncTxProcessing = ConfigService.get('USE_ASYNC_TX_PROCESSING');
     if (useAsyncTxProcessing) {
-      this.sendRawTransactionProcessor(
-        transactionBuffer,
-        parsedTx,
-        networkGasPriceInWeiBars,
-        this.rawTxSynchronizeService,
-        requestDetails,
-      );
+      this.sendRawTransactionProcessor(transactionBuffer, parsedTx, networkGasPriceInWeiBars, requestDetails);
       return Utils.computeTransactionHash(transactionBuffer);
     }
 
@@ -293,7 +272,6 @@ export class TransactionService implements ITransactionService {
       transactionBuffer,
       parsedTx,
       networkGasPriceInWeiBars,
-      this.rawTxSynchronizeService,
       requestDetails,
     );
   }
@@ -491,7 +469,6 @@ export class TransactionService implements ITransactionService {
     transactionBuffer: Buffer,
     parsedTx: EthersTransaction,
     networkGasPriceInWeiBars: number,
-    rawTxSynchronizeService: RawTxSynchronizeService,
     requestDetails: RequestDetails,
   ): Promise<string | JsonRpcError> {
     let sendRawTransactionError: any;
@@ -506,7 +483,6 @@ export class TransactionService implements ITransactionService {
       transactionBuffer,
       originalCallerAddress,
       networkGasPriceInWeiBars,
-      rawTxSynchronizeService,
       requestDetails,
     );
 
@@ -660,7 +636,6 @@ export class TransactionService implements ITransactionService {
     transactionBuffer: Buffer,
     originalCallerAddress: string,
     networkGasPriceInWeiBars: number,
-    rawTxSynchronizeService: RawTxSynchronizeService,
     requestDetails: RequestDetails,
   ): Promise<{
     txSubmitted: boolean;
@@ -680,7 +655,6 @@ export class TransactionService implements ITransactionService {
         originalCallerAddress,
         networkGasPriceInWeiBars,
         await this.getCurrentNetworkExchangeRateInCents(requestDetails),
-        rawTxSynchronizeService,
       );
 
       txSubmitted = true;

packages/relay/src/lib/services/hapiService/hapiService.ts

@@ -9,7 +9,6 @@ import { Counter, Registry } from 'prom-client';
 import { SDKClient } from '../../clients';
 import { ITransactionRecordMetric, RequestDetails, TypedEvents } from '../../types';
 import { HbarLimitService } from '../hbarLimitService';
-import { RawTxSynchronizeService } from '../rawTxSynchronizeService/RawTxSynchronizeService';
 
 export default class HAPIService {
   /**
@@ -243,7 +242,6 @@ export default class HAPIService {
     originalCallerAddress: string,
     networkGasPriceInWeiBars: number,
     currentNetworkExchangeRateInCents: number,
-    rawTxSynchronizeService: RawTxSynchronizeService,
   ): Promise<{ txResponse: TransactionResponse; fileId: FileId | null }> {
     return this.getSDKClient().submitEthereumTransaction(
       transactionBuffer,
@@ -252,7 +250,6 @@ export default class HAPIService {
       originalCallerAddress,
       networkGasPriceInWeiBars,
       currentNetworkExchangeRateInCents,
-      rawTxSynchronizeService,
     );
   }
 

packages/relay/src/lib/services/rawTxSynchronizeService/RawTxSynchronizeService.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { Mutex, withTimeout } from 'async-mutex';
+import { Transaction } from 'ethers';
 import { LRUCache } from 'lru-cache';
 import { Logger } from 'pino';
 
@@ -49,62 +50,54 @@ export class RawTxSynchronizeService {
   }
 
   /**
-   * Acquires a mutex lock for the specified sender address with timeout.
+   * Executes a callback under an exclusive per-sender mutex with timeout protection.
    *
-   * This method implements timeout-based lock acquisition. The call will wait for the mutex
-   * to become available but will timeout after the configured duration to prevent
-   * deadlocks and ensure resource availability.
+   * The lock is automatically released once the callback settles (resolve or reject). If the lock cannot be
+   * acquired within the configured timeout, an error is thrown mirroring the previous behaviour.
    *
-   * @param sender - The sender address (wallet address) to acquire the lock for
-   * @returns Promise that resolves when the lock is successfully acquired
-   * @throws Error if the lock acquisition fails due to timeout or internal errors
+   * @param rawTransaction - The raw transaction payload used to derive the sender address
+   * @param callback - The critical section to execute while the sender lock is held
+   * @returns The callback result
    */
-  async acquireLock(sender: string): Promise<void> {
-    const mutex = this.getOrCreateMutex(sender);
+  async runExclusive<T>(rawTransaction: string, callback: () => Promise<T> | T): Promise<T> {
+    const senderAddress = this.extractSender(rawTransaction);
+
+    if (!senderAddress) {
+      this.logger.warn('Unable to derive sender from raw transaction. Executing callback without synchronization.');
+      return await callback();
+    }
+
+    const mutex = this.getOrCreateMutex(senderAddress);
     const timeoutMutex = withTimeout(mutex, RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS);
+    const waitStartedAt = Date.now();
+    let lockAcquired = false;
 
     try {
-      // Acquire the mutex lock with timeout protection
-      await timeoutMutex.acquire();
-      this.logger.debug(
-        `Lock acquired for sender: ${sender}, timeout: ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms`,
-      );
+      return await timeoutMutex.runExclusive(async () => {
+        const waitDurationMs = Date.now() - waitStartedAt;
+        this.logger.debug(
+          `Lock acquired for sender: ${senderAddress}, waited ${waitDurationMs}ms (timeout: ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms)`,
+        );
+        lockAcquired = true;
+
+        return await callback();
+      });
     } catch (error) {
       if (error instanceof Error && error.message.includes('timeout')) {
         throw new Error(
-          `Failed to acquire lock for sender ${sender}: timeout after ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms`,
+          `Failed to acquire lock for sender ${senderAddress}: timeout after ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms`,
         );
       }
-      this.logger.error(`Failed to acquire lock for ${sender}:`, error);
-      throw error;
-    }
-  }
-
-  /**
-   * Releases the mutex lock for the specified sender address.
-   *
-   * This method safely releases a previously acquired lock using the mutex instance and performs cleanup
-   * to prevent resource leaks. It's safe to call even if no lock is held for the sender.
-   *
-   * @param sender - The sender address to release the lock for
-   * @returns Promise that resolves when the lock is successfully released
-   */
-  async releaseLock(sender: string): Promise<void> {
-    const mutex = this.localLockStates.get(sender);
-    if (!mutex || !mutex.isLocked()) {
-      this.logger.debug(`No active lock to release for sender: ${sender}`);
-      return;
-    }
 
-    try {
-      mutex.release();
-      this.logger.debug(`Lock released for sender: ${sender}`);
-    } catch (error) {
-      this.logger.error(`Error releasing lock for ${sender}:`, error);
+      this.logger.error(`Failed to execute exclusive section for ${senderAddress}:`, error);
+      throw error;
+    } finally {
+      // only log release event if lock was actually acquired
+      if (lockAcquired) {
+        this.logger.debug(`Lock released for sender: ${senderAddress}`);
+      }
     }
-  }
-
-  /**
+  } /**
    * Retrieves an existing mutex for the sender or creates a new one if needed.
    *
    * @private
@@ -119,4 +112,18 @@ export class RawTxSynchronizeService {
     }
     return mutex;
   }
+
+  /**
+   * Parses the raw transaction and returns the normalized sender address if available.
+   *
+   * @param rawTransaction - The serialized transaction payload (with or without 0x prefix)
+   * @returns The lowercase sender address, or null when parsing fails or sender is absent
+   */
+  private extractSender(rawTransaction: string): string | null | undefined {
+    try {
+      return Transaction.from(rawTransaction).from?.toLowerCase();
+    } catch {
+      return null;
+    }
+  }
 }