Add TransactionManager to handle stalled tx

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):

src/common/tests/test_transaction.py

@@ -0,0 +1,171 @@
+from math import ceil
+from unittest import mock
+
+import pytest
+from hexbytes import HexBytes
+from web3 import Web3
+from web3.types import Wei
+
+from src.common.transaction import REPLACEMENT_GAS_BUMP, TransactionManager
+
+GWEI = Web3.to_wei(1, 'gwei')
+
+
+@pytest.mark.usefixtures('fake_settings')
+class TestTransactionManager:
+    async def test_no_pending_high_priority_uses_latest_nonce(self):
+        transact = mock.AsyncMock(return_value=HexBytes('0x01'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=5, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al:
+            manager = TransactionManager()
+            receipt = await manager.transact(_tx_function(transact), high_priority=True)
+
+        assert receipt is not None
+        params = transact.call_args.args[0]
+        assert params['nonce'] == 5
+        assert params['maxFeePerGas'] == GWEI
+        assert params['maxPriorityFeePerGas'] == GWEI // 2
+
+    async def test_default_gas_skips_fee_fields(self):
+        # high_priority=False with no pending tx submits with the node's default gas
+        transact = mock.AsyncMock(return_value=HexBytes('0x01'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=5, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al:
+            manager = TransactionManager()
+            await manager.transact(_tx_function(transact))
+
+        params = transact.call_args.args[0]
+        assert params['nonce'] == 5
+        assert 'maxFeePerGas' not in params
+        assert 'maxPriorityFeePerGas' not in params
+
+    async def test_default_gas_escalates_on_fee_too_low(self):
+        fee_too_low = ValueError({'code': -32010})
+        # every default-gas attempt is rejected, the final escalation succeeds
+        transact = mock.AsyncMock(
+            side_effect=[fee_too_low, fee_too_low, fee_too_low, HexBytes('0x02')]
+        )
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=5, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al, mock.patch(
+            'src.common.transaction.ATTEMPTS_WITH_DEFAULT_GAS', 3
+        ), mock.patch('src.common.transaction.asyncio.sleep', mock.AsyncMock()):
+            manager = TransactionManager()
+            receipt = await manager.transact(_tx_function(transact))
+
+        assert receipt is not None
+        # 3 default-gas attempts + 1 high-priority escalation
+        assert transact.await_count == 4
+        escalated = transact.call_args_list[-1].args[0]
+        assert escalated['maxFeePerGas'] == GWEI
+        assert escalated['maxPriorityFeePerGas'] == GWEI // 2
+
+    async def test_pending_reuses_nonce_and_bumps_gas(self):
+        manager = TransactionManager()
+
+        # first submission records the gas used for nonce 5
+        transact1 = mock.AsyncMock(return_value=HexBytes('0x01'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=5, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al:
+            await manager.transact(_tx_function(transact1), high_priority=True)
+
+        # second submission sees a pending tx at nonce 5 -> replace it, bumped
+        transact2 = mock.AsyncMock(return_value=HexBytes('0x02'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=6, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al:
+            await manager.transact(_tx_function(transact2))
+
+        params = transact2.call_args.args[0]
+        assert params['nonce'] == 5  # same nonce, not 6
+        assert params['maxFeePerGas'] == ceil(GWEI * REPLACEMENT_GAS_BUMP)
+        assert params['maxPriorityFeePerGas'] == ceil((GWEI // 2) * REPLACEMENT_GAS_BUMP)
+
+    async def test_pending_skips_default_gas(self):
+        # even without high_priority, a pending tx forces the high-priority path
+        transact = mock.AsyncMock(return_value=HexBytes('0x02'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=6, gas_manager=_gas_manager(GWEI, GWEI // 2)
+        )
+        with ec, w, gm, al:
+            manager = TransactionManager()
+            await manager.transact(_tx_function(transact))
+
+        assert transact.await_count == 1
+        params = transact.call_args.args[0]
+        assert params['nonce'] == 5
+        assert params['maxFeePerGas'] == GWEI
+
+    async def test_reverted_receipt_returns_none(self):
+        transact = mock.AsyncMock(return_value=HexBytes('0x01'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5,
+            pending_nonce=5,
+            gas_manager=_gas_manager(GWEI, GWEI // 2),
+            status=0,
+        )
+        with ec, w, gm, al:
+            manager = TransactionManager()
+            receipt = await manager.transact(_tx_function(transact), high_priority=True)
+
+        assert receipt is None
+
+    async def test_gas_capped_at_max_fee_per_gas(self):
+        # high-priority returns far above the HOODI ceiling (10 gwei)
+        huge = Web3.to_wei(100, 'gwei')
+        cap = Web3.to_wei(10, 'gwei')
+        transact = mock.AsyncMock(return_value=HexBytes('0x01'))
+        ec, w, gm, al = _patch(
+            latest_nonce=5, pending_nonce=5, gas_manager=_gas_manager(huge, huge)
+        )
+        with ec, w, gm, al:
+            manager = TransactionManager()
+            await manager.transact(_tx_function(transact), high_priority=True)
+
+        params = transact.call_args.args[0]
+        assert params['maxFeePerGas'] == cap
+        assert params['maxPriorityFeePerGas'] <= params['maxFeePerGas']
+
+
+def _gas_manager(max_fee: int, priority_fee: int) -> mock.Mock:
+    manager = mock.Mock()
+    manager.get_high_priority_tx_params = mock.AsyncMock(
+        return_value={'maxFeePerGas': Wei(max_fee), 'maxPriorityFeePerGas': Wei(priority_fee)}
+    )
+    return manager
+
+
+def _patch(latest_nonce: int, pending_nonce: int, gas_manager: mock.Mock, status: int = 1):
+    execution_client = mock.Mock()
+    execution_client.eth.get_transaction_count = mock.AsyncMock(
+        side_effect=[latest_nonce, pending_nonce]
+    )
+    execution_client.eth.wait_for_transaction_receipt = mock.AsyncMock(
+        return_value={
+            'status': status,
+            'transactionHash': HexBytes('0xab'),
+            'blockNumber': 1,
+        }
+    )
+    wallet = mock.Mock()
+    wallet.address = '0x' + '11' * 20
+    return (
+        mock.patch('src.common.transaction.execution_client', execution_client),
+        mock.patch('src.common.transaction.wallet', wallet),
+        mock.patch('src.common.transaction.build_gas_manager', return_value=gas_manager),
+        mock.patch('src.common.transaction.is_alchemy_used', return_value=False),
+    )
+
+
+def _tx_function(transact_mock: mock.AsyncMock) -> mock.Mock:
+    tx_function = mock.Mock()
+    tx_function.transact = transact_mock
+    return tx_function