Add TransactionManager to handle stalled tx (#767)

* Replace stuck pending transactions instead of queuing new nonce

* Integrate transaction_gas_wrapper into TransactionManager

* Review fix

* Handle pending tx timeout and centralize estimate_gas

* Add test for pending tx receipt timeout

* Skip replacement when pending tx is at fee ceiling

* Skip replacement when fee bump cannot clear node threshold

* Add Fees unit tests and rename to_tx_params

* Detect fee-too-low rejections via Web3RPCError message

* Match Erigon replacement rejection message

* Decrease EXECUTION_TRANSACTION_TIMEOUT default to 60 sec

* Simplify default-gas escalation branch

Author: evgeny-stakewise

src/common/contracts.py

@@ -14,10 +14,10 @@
     AsyncContractEvents,
     AsyncContractFunctions,
 )
-from web3.types import BlockNumber, ChecksumAddress, EventData, Wei
+from web3.types import BlockNumber, ChecksumAddress, EventData, TxReceipt, Wei
 
 from src.common.clients import execution_client as default_execution_client
-from src.common.execution import transaction_gas_wrapper
+from src.common.transaction import tx_manager
 from src.common.typings import (
     ExitQueueMissingAssetsParams,
     HarvestParams,
@@ -411,10 +411,9 @@ async def aggregate(
     async def tx_aggregate(
         self,
         data: list[tuple[ChecksumAddress, HexStr]],
-    ) -> HexStr:
+    ) -> TxReceipt | None:
         tx_function = self.contract.functions.aggregate(data)
-        tx_hash = await transaction_gas_wrapper(tx_function)
-        return Web3.to_hex(tx_hash)
+        return await tx_manager.transact(tx_function)
 
 
 class ValidatorsCheckerContract(ContractWrapper):

src/common/execution.py

@@ -1,19 +1,16 @@
-import asyncio
 import logging
 from urllib.parse import urlparse
 
-from hexbytes import HexBytes
 from sw_utils import GasManager, InterruptHandler
 from web3 import Web3
-from web3.contract.async_contract import AsyncContractFunction
-from web3.types import TxParams, Wei
+from web3.types import Wei
 
 from src.common.clients import execution_client
 from src.common.metrics import metrics
 from src.common.tasks import BaseTask
 from src.common.wallet import wallet
 from src.config.networks import HOODI
-from src.config.settings import ATTEMPTS_WITH_DEFAULT_GAS, settings
+from src.config.settings import settings
 
 logger = logging.getLogger(__name__)
 
@@ -49,40 +46,10 @@ async def get_wallet_balance() -> Wei:
     return await execution_client.eth.get_balance(wallet.address)
 
 
-async def transaction_gas_wrapper(
-    tx_function: AsyncContractFunction, tx_params: TxParams | None = None
-) -> HexBytes:
-    """Handles periods with high gas in the network."""
-    if not tx_params:
-        tx_params = {}
-
-    # trying to submit with basic gas
-    attempts_with_default_gas = ATTEMPTS_WITH_DEFAULT_GAS
-
-    # Alchemy does not support eth_maxPriorityFeePerGas for Hoodi
-    if settings.network == HOODI and _is_alchemy_used():
-        attempts_with_default_gas = 0
-
-    for i in range(attempts_with_default_gas):
-        try:
-            return await tx_function.transact(tx_params)
-        except ValueError as e:
-            # Handle only FeeTooLow error
-            if not _is_fee_too_low_error(e):
-                raise e
-            if i < attempts_with_default_gas - 1:  # skip last sleep
-                await asyncio.sleep(settings.network_config.SECONDS_PER_BLOCK)
-
-    # use high priority fee
-    gas_manager = build_gas_manager()
-    tx_params = tx_params | await gas_manager.get_high_priority_tx_params()
-    return await tx_function.transact(tx_params)
-
-
 async def check_gas_price(high_priority: bool = False) -> bool:
     gas_manager = build_gas_manager()
     # Alchemy does not support eth_maxPriorityFeePerGas for Hoodi, skip
-    if settings.network == HOODI and _is_alchemy_used():
+    if settings.network == HOODI and is_alchemy_used():
         return True
 
     return await gas_manager.check_gas_price(high_priority)
@@ -99,14 +66,7 @@ def build_gas_manager() -> GasManager:
     )
 
 
-def _is_fee_too_low_error(e: ValueError) -> bool:
-    code = None
-    if e.args and isinstance(e.args[0], dict):
-        code = e.args[0].get('code')
-    return code == -32010
-
-
-def _is_alchemy_used() -> bool:
+def is_alchemy_used() -> bool:
     for endpoint in settings.execution_endpoints:
         domain = urlparse(endpoint).netloc
         if domain.lower().endswith(ALCHEMY_DOMAIN):