feat: introduced session key

Signed-off-by: Logan Nguyen <[email protected]>

Author: quiet-node

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

@@ -135,6 +135,8 @@ export class SDKClient {
    * @param {string} originalCallerAddress - The address of the original caller making the request.
    * @param {number} networkGasPriceInWeiBars - The predefined gas price of the network in weibar.
    * @param {number} currentNetworkExchangeRateInCents - The exchange rate in cents of the current network.
+   * @param {RawTxSynchronizeService} rawTxSynchronizeService - The service for managing transaction locks.
+   * @param {string | null} lockSessionKey - The session key for the acquired lock, null if no lock was acquired.
    * @returns {Promise<{ txResponse: TransactionResponse; fileId: FileId | null }>}
    * @throws {SDKClientError} Throws an error if no file ID is created or if the preemptive fee check fails.
    */
@@ -146,6 +148,7 @@ export class SDKClient {
     networkGasPriceInWeiBars: number,
     currentNetworkExchangeRateInCents: number,
     rawTxSynchronizeService: RawTxSynchronizeService,
+    lockSessionKey: string | null,
   ): Promise<{ txResponse: TransactionResponse; fileId: FileId | null }> {
     const jumboTxEnabled = ConfigService.get('JUMBO_TX_ENABLED');
     const ethereumTransactionData: EthereumTransactionData = EthereumTransactionData.fromBytes(transactionBuffer);
@@ -192,6 +195,7 @@ export class SDKClient {
         true,
         originalCallerAddress,
         rawTxSynchronizeService,
+        lockSessionKey,
       ),
     };
   }
@@ -268,7 +272,9 @@ export class SDKClient {
    * @param requestDetails - The request details for logging and tracking.
    * @param shouldThrowHbarLimit - Flag to indicate whether to check HBAR limits.
    * @param originalCallerAddress - The address of the original caller making the request.
+   * @param rawTxSynchronizeService - The service for managing transaction locks.
    * @param estimatedTxFee - The optional total estimated transaction fee.
+   * @param lockSessionKey - The session key for the acquired lock, null if no lock was acquired.
    * @returns - A promise that resolves to the transaction response.
    * @throws {SDKClientError} - Throws if an error occurs during transaction execution.
    */
@@ -279,6 +285,7 @@ export class SDKClient {
     shouldThrowHbarLimit: boolean,
     originalCallerAddress: string,
     rawTxSynchronizeService?: RawTxSynchronizeService,
+    lockSessionKey?: string | null,
     estimatedTxFee?: number,
   ): Promise<TransactionResponse> {
     const txConstructorName = transaction.constructor.name;
@@ -337,8 +344,8 @@ 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 (rawTxSynchronizeService && lockSessionKey) {
+        await rawTxSynchronizeService.releaseLock(originalCallerAddress, lockSessionKey);
       }
 
       if (transactionId?.length) {

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

@@ -256,46 +256,57 @@ export class TransactionService implements ITransactionService {
 
     const transactionBuffer = Buffer.from(this.prune0x(transaction), 'hex');
     const parsedTx = Precheck.parseRawTransaction(transaction);
+    let lockSessionKey: string | null = null;
 
     // 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);
+      lockSessionKey = await this.rawTxSynchronizeService.acquireLock(parsedTx.from);
     }
 
-    const networkGasPriceInWeiBars = Utils.addPercentageBufferToGasPrice(
-      await this.common.getGasPriceInWeibars(requestDetails),
-    );
+    try {
+      const networkGasPriceInWeiBars = Utils.addPercentageBufferToGasPrice(
+        await this.common.getGasPriceInWeibars(requestDetails),
+      );
 
-    await this.validateRawTransaction(parsedTx, networkGasPriceInWeiBars, requestDetails);
+      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(
+      /**
+       * 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,
+          this.rawTxSynchronizeService,
+          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,
         this.rawTxSynchronizeService,
+        lockSessionKey,
         requestDetails,
       );
-      return Utils.computeTransactionHash(transactionBuffer);
+    } catch (error) {
+      // Release the lock on any error to prevent lock starvation
+      if (lockSessionKey && parsedTx.from) {
+        await this.rawTxSynchronizeService.releaseLock(parsedTx.from, lockSessionKey);
+      }
+      throw error;
     }
-
-    /**
-     * 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,
-      this.rawTxSynchronizeService,
-      requestDetails,
-    );
   }
 
   /**
@@ -484,6 +495,8 @@ export class TransactionService implements ITransactionService {
    * @param {Buffer} transactionBuffer - The raw transaction data as a buffer.
    * @param {EthersTransaction} parsedTx - The parsed Ethereum transaction object.
    * @param {number} networkGasPriceInWeiBars - The current network gas price in wei bars.
+   * @param {RawTxSynchronizeService} rawTxSynchronizeService - The service for managing transaction locks.
+   * @param {string | null} lockSessionKey - The session key for the acquired lock, null if no lock was acquired.
    * @param {RequestDetails} requestDetails - Details of the request for logging and tracking purposes.
    * @returns {Promise<string | JsonRpcError>} A promise that resolves to the transaction hash if successful, or a JsonRpcError if an error occurs.
    */
@@ -492,6 +505,7 @@ export class TransactionService implements ITransactionService {
     parsedTx: EthersTransaction,
     networkGasPriceInWeiBars: number,
     rawTxSynchronizeService: RawTxSynchronizeService,
+    lockSessionKey: string | null,
     requestDetails: RequestDetails,
   ): Promise<string | JsonRpcError> {
     let sendRawTransactionError: any;
@@ -507,6 +521,7 @@ export class TransactionService implements ITransactionService {
       originalCallerAddress,
       networkGasPriceInWeiBars,
       rawTxSynchronizeService,
+      lockSessionKey,
       requestDetails,
     );
 
@@ -653,6 +668,8 @@ export class TransactionService implements ITransactionService {
    * @param transactionBuffer The raw transaction buffer
    * @param originalCallerAddress The address of the original caller
    * @param networkGasPriceInWeiBars The current network gas price in wei bars
+   * @param rawTxSynchronizeService The service for managing transaction locks
+   * @param lockSessionKey The session key for the acquired lock, null if no lock was acquired
    * @param requestDetails The request details for logging and tracking
    * @returns {Promise<{txSubmitted: boolean, submittedTransactionId: string, error: any}>} A promise that resolves to an object containing transaction submission details
    */
@@ -661,6 +678,7 @@ export class TransactionService implements ITransactionService {
     originalCallerAddress: string,
     networkGasPriceInWeiBars: number,
     rawTxSynchronizeService: RawTxSynchronizeService,
+    lockSessionKey: string | null,
     requestDetails: RequestDetails,
   ): Promise<{
     txSubmitted: boolean;
@@ -681,6 +699,7 @@ export class TransactionService implements ITransactionService {
         networkGasPriceInWeiBars,
         await this.getCurrentNetworkExchangeRateInCents(requestDetails),
         rawTxSynchronizeService,
+        lockSessionKey,
       );
 
       txSubmitted = true;

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

@@ -244,6 +244,7 @@ export default class HAPIService {
     networkGasPriceInWeiBars: number,
     currentNetworkExchangeRateInCents: number,
     rawTxSynchronizeService: RawTxSynchronizeService,
+    lockSessionKey: string | null,
   ): Promise<{ txResponse: TransactionResponse; fileId: FileId | null }> {
     return this.getSDKClient().submitEthereumTransaction(
       transactionBuffer,
@@ -253,6 +254,7 @@ export default class HAPIService {
       networkGasPriceInWeiBars,
       currentNetworkExchangeRateInCents,
       rawTxSynchronizeService,
+      lockSessionKey,
     );
   }
 

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

@@ -1,21 +1,33 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { Mutex, withTimeout } from 'async-mutex';
+import { randomUUID } from 'crypto';
 import { LRUCache } from 'lru-cache';
 import { Logger } from 'pino';
 
+/**
+ * Represents the state of a lock for a specific sender address.
+ * Encapsulates both the mutex and the active session keys for that lock.
+ */
+interface LockState {
+  /** The mutex used for synchronization */
+  mutex: Mutex;
+  /** Set of active session keys that can release this lock */
+  activeSessionKeys: Set<string>;
+}
+
 export class RawTxSynchronizeService {
-  /** Default timeout for lock acquisition in milliseconds (30 seconds) */
-  private static readonly DEFAULT_LOCK_TIMEOUT_MS = 30_000;
+  /** Default timeout for lock acquisition in milliseconds (300 seconds) */
+  private static readonly DEFAULT_LOCK_TIMEOUT_MS = 300_000;
 
   /** Default TTL for inactive sender mutex entries in milliseconds (15 minutes) */
   private static readonly DEFAULT_LOCK_STATE_TTL_MS = 15 * 60 * 1000;
 
   /** Maximum number of sender mutex entries to maintain concurrently */
   private static readonly MAX_LOCKS = 1000;
 
-  /** LRU cache with TTL for managing sender mutexes with automatic cleanup */
-  private readonly localLockStates: LRUCache<string, Mutex>;
+  /** LRU cache with TTL for managing sender lock states with automatic cleanup */
+  private readonly localLockStates: LRUCache<string, LockState>;
 
   /** Logger instance for debugging and monitoring lock operations */
   private readonly logger: Logger;
@@ -30,19 +42,21 @@ export class RawTxSynchronizeService {
    */
   constructor(logger: Logger) {
     this.logger = logger;
-    this.localLockStates = new LRUCache<string, Mutex>({
+    this.localLockStates = new LRUCache<string, LockState>({
       max: RawTxSynchronizeService.MAX_LOCKS,
       ttl: RawTxSynchronizeService.DEFAULT_LOCK_STATE_TTL_MS,
-      dispose: (mutex: Mutex, sender: string) => {
+      dispose: (lockState: LockState, sender: string) => {
         // Clean up any active locks during eviction/expiration
-        if (mutex.isLocked()) {
+        if (lockState.mutex.isLocked()) {
           try {
-            mutex.release();
+            lockState.mutex.release();
             this.logger.debug(`Active lock auto-released during cleanup for sender: ${sender}`);
           } catch (error) {
             this.logger.warn(`Error auto-releasing lock during cleanup for sender: ${sender}`, error);
           }
         }
+        // Clear all active sessions since the lock state is being disposed
+        lockState.activeSessionKeys.clear();
         this.logger.debug(`Lock state evicted/expired for sender: ${sender}`);
       },
     });
@@ -51,26 +65,32 @@ export class RawTxSynchronizeService {
   /**
    * Acquires a mutex lock for the specified sender address with timeout.
    *
-   * 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.
+   * This method implements timeout-based lock acquisition and generates a unique session key
+   * to track the lock instance. The session key can be used to release the specific lock
+   * and prevent double-release scenarios.
    *
    * @param sender - The sender address (wallet address) to acquire the lock for
-   * @returns Promise that resolves when the lock is successfully acquired
+   * @returns Promise that resolves to a unique session key when the lock is successfully acquired
    * @throws Error if the lock acquisition fails due to timeout or internal errors
    */
-  async acquireLock(sender: string): Promise<void> {
-    const mutex = this.getOrCreateMutex(sender);
-    const timeoutMutex = withTimeout(mutex, RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS);
+  async acquireLock(sender: string): Promise<string> {
+    const lockState = this.getOrCreateLockState(sender);
+    const timeoutMutex = withTimeout(lockState.mutex, RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS);
     const waitStartedAt = Date.now();
+    const sessionKey = randomUUID();
 
     try {
       // Acquire the mutex lock with timeout protection
       await timeoutMutex.acquire();
+
+      // Add the session key to the active sessions set
+      lockState.activeSessionKeys.add(sessionKey);
+
       const waitDurationMs = Date.now() - waitStartedAt;
       this.logger.debug(
-        `Lock acquired for sender: ${sender}, waited ${waitDurationMs}ms (timeout: ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms)`,
+        `Lock acquired for sender: ${sender}, sessionKey: ${sessionKey}, waited ${waitDurationMs}ms (timeout: ${RawTxSynchronizeService.DEFAULT_LOCK_TIMEOUT_MS}ms)`,
       );
+      return sessionKey;
     } catch (error) {
       if (error instanceof Error && error.message.includes('timeout')) {
         throw new Error(
@@ -83,42 +103,50 @@ export class RawTxSynchronizeService {
   }
 
   /**
-   * Releases the mutex lock for the specified sender address.
+   * Releases the mutex lock for the specified sender address using a session key.
    *
-   * 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.
+   * This method safely releases a previously acquired lock by checking if the session key
+   * is in the active sessions set. If the session key exists, it releases the mutex and
+   * removes the session key. If not, it means the lock for that session was already released (double-release protection).
    *
    * @param sender - The sender address to release the lock for
-   * @returns Promise that resolves when the lock is successfully released
+   * @param sessionKey - The unique session key returned from acquireLock()
+   * @returns Promise that resolves when the lock is successfully released or if already 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}`);
+  async releaseLock(sender: string, sessionKey: string): Promise<void> {
+    const lockState = this.localLockStates.get(sender);
+
+    if (!lockState || !lockState.activeSessionKeys.has(sessionKey)) {
+      this.logger.debug(`Lock already released or not found for sender: ${sender}, session: ${sessionKey}`);
       return;
     }
 
     try {
-      mutex.release();
-      this.logger.debug(`Lock released for sender: ${sender}`);
+      lockState.mutex.release();
+      this.logger.debug(`Lock released for sender: ${sender}, session: ${sessionKey}`);
     } catch (error) {
-      this.logger.error(`Error releasing lock for ${sender}:`, error);
+      this.logger.error(`Error releasing lock for ${sender}, session: ${sessionKey}:`, error);
+    } finally {
+      // Always remove the session key to prevent stale state, regardless of success or error
+      lockState.activeSessionKeys.delete(sessionKey);
     }
   }
-
   /**
-   * Retrieves an existing mutex for the sender or creates a new one if needed.
+   * Retrieves an existing lock state for the sender or creates a new one if needed.
    *
    * @private
-   * @param sender - The sender address to get or create a mutex for
-   * @returns The mutex associated with the specified sender
+   * @param sender - The sender address to get or create a lock state for
+   * @returns The lock state associated with the specified sender
    */
-  private getOrCreateMutex(sender: string): Mutex {
-    let mutex = this.localLockStates.get(sender);
-    if (!mutex) {
-      mutex = new Mutex();
-      this.localLockStates.set(sender, mutex);
+  private getOrCreateLockState(sender: string): LockState {
+    let lockState = this.localLockStates.get(sender);
+    if (!lockState) {
+      lockState = {
+        mutex: new Mutex(),
+        activeSessionKeys: new Set<string>(),
+      };
+      this.localLockStates.set(sender, lockState);
     }
-    return mutex;
+    return lockState;
   }
 }