Fix not retrying the previous attempt when failing connectivity check (…

…#8651)

* Fix not retrying the previous attempt when failing connectivity check

* Move connectivity error to gas models

core/chains/evm/gas/block_history_estimator.go

@@ -73,7 +73,6 @@ var (
 
 const BumpingHaltedLabel = "Tx gas bumping halted since price exceeds current block prices by significant margin; tx will continue to be rebroadcasted but your node, RPC, or the chain might be experiencing connectivity issues; please investigate and fix ASAP"
 
-var ErrConnectivity = errors.New("transaction propagation issue: transactions are not being mined")
 
 var _ EvmEstimator = &BlockHistoryEstimator{}
 

core/chains/evm/gas/block_history_estimator_test.go

@@ -1997,6 +1997,7 @@ func TestBlockHistoryEstimator_CheckConnectivity(t *testing.T) {
 			err := bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has gas price of 7 wei, which is above percentile=40%% (percentile price: 5 wei) for blocks 2 thru 0 (checking 3 blocks)", attempts[3].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 		})
 
 		t.Run("fails check if one or more blocks has percentile price higher than any transaction gas price", func(t *testing.T) {
@@ -2013,6 +2014,7 @@ func TestBlockHistoryEstimator_CheckConnectivity(t *testing.T) {
 			err = bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has gas price of 3 wei, which is above percentile=5%% (percentile price: 2 wei) for blocks 2 thru 0 (checking 3 blocks)", attempts[1].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 		})
 	})
 
@@ -2044,6 +2046,7 @@ func TestBlockHistoryEstimator_CheckConnectivity(t *testing.T) {
 			err := bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has gas price of 10 wei, which is above percentile=60%% (percentile price: 7 wei) for blocks 3 thru 3 (checking 1 blocks)", attempts[1].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 		})
 
 		attempts = []txmgrtypes.PriorAttempt[gas.EvmFee, common.Hash]{
@@ -2059,6 +2062,7 @@ func TestBlockHistoryEstimator_CheckConnectivity(t *testing.T) {
 			err := bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has tip cap of 10 wei, which is above percentile=60%% (percentile tip cap: 6 wei) for blocks 3 thru 3 (checking 1 blocks)", attempts[0].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 		})
 
 	})
@@ -2116,13 +2120,15 @@ func TestBlockHistoryEstimator_CheckConnectivity(t *testing.T) {
 			err := bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has tip cap of 5 wei, which is above percentile=20%% (percentile tip cap: 4 wei) for blocks 2 thru 0 (checking 3 blocks)", attempts[2].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 
 			cfg.BlockHistoryEstimatorCheckInclusionBlocksF = 3
 			cfg.BlockHistoryEstimatorCheckInclusionPercentileF = 5
 
 			err = bhe.CheckConnectivity(attempts)
 			require.Error(t, err)
 			assert.Contains(t, err.Error(), fmt.Sprintf("transaction %s has tip cap of 3 wei, which is above percentile=5%% (percentile tip cap: 2 wei) for blocks 2 thru 0 (checking 3 blocks)", attempts[1].GetHash()))
+			require.ErrorIs(t, err, gas.ErrConnectivity)
 		})
 
 		t.Run("passes check if, for at least one block, feecap < tipcap+basefee, even if percentile is not reached", func(t *testing.T) {

core/chains/evm/gas/models.go

@@ -22,10 +22,11 @@ import (
 var (
 	ErrBumpGasExceedsLimit = errors.New("gas bump exceeds limit")
 	ErrBump                = errors.New("gas bump failed")
+	ErrConnectivity        = errors.New("transaction propagation issue: transactions are not being mined")
 )
 
 func IsBumpErr(err error) bool {
-	return err != nil && (errors.Is(err, ErrBumpGasExceedsLimit) || errors.Is(err, ErrBump))
+	return err != nil && (errors.Is(err, ErrBumpGasExceedsLimit) || errors.Is(err, ErrBump) || errors.Is(err, ErrConnectivity))
 }
 
 // NewEstimator returns the estimator for a given config

core/chains/evm/txmgr/eth_confirmer_test.go

@@ -9,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	pkgerrors "github.com/pkg/errors"
+
 	gethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -21,6 +23,8 @@ import (
 	"github.com/smartcontractkit/chainlink/core/assets"
 	evmclient "github.com/smartcontractkit/chainlink/core/chains/evm/client"
 	evmconfig "github.com/smartcontractkit/chainlink/core/chains/evm/config"
+	"github.com/smartcontractkit/chainlink/core/chains/evm/gas"
+	gasmocks "github.com/smartcontractkit/chainlink/core/chains/evm/gas/mocks"
 	"github.com/smartcontractkit/chainlink/core/chains/evm/txmgr"
 	evmtypes "github.com/smartcontractkit/chainlink/core/chains/evm/types"
 	"github.com/smartcontractkit/chainlink/core/internal/cltest"
@@ -1534,6 +1538,92 @@ func TestEthConfirmer_FindEthTxsRequiringRebroadcast(t *testing.T) {
 	})
 }
 
+func TestEthConfirmer_RebroadcastWhereNecessary_WithConnectivityCheck(t *testing.T) {
+	t.Parallel()
+	lggr := logger.TestLogger(t)
+
+	db := pgtest.NewSqlxDB(t)
+	ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+
+	t.Run("should retry previous attempt if connectivity check failed for legacy transactions", func(t *testing.T) {
+		cfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+			c.EVM[0].GasEstimator.EIP1559DynamicFees = ptr(false)
+			c.EVM[0].GasEstimator.BlockHistory.BlockHistorySize = ptr[uint16](2)
+			c.EVM[0].GasEstimator.BlockHistory.CheckInclusionBlocks = ptr[uint16](4)
+		})
+		evmcfg := evmtest.NewChainScopedConfig(t, cfg)
+
+		borm := cltest.NewTxmORM(t, db, cfg)
+		ethKeyStore := cltest.NewKeyStore(t, db, cfg).Eth()
+		state, fromAddress := cltest.MustInsertRandomKeyReturningState(t, ethKeyStore)
+		keys := []ethkey.State{state}
+		kst := ksmocks.NewEth(t)
+
+		estimator := gasmocks.NewEvmEstimator(t)
+		estimator.On("BumpLegacyGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, uint32(0), pkgerrors.Wrapf(gas.ErrConnectivity, "transaction..."))
+		// Create confirmer with necessary state
+		ec := txmgr.NewEthConfirmer(borm, ethClient, evmcfg, kst, keys, gas.NewWrappedEvmEstimator(estimator, evmcfg), nil, lggr)
+		currentHead := int64(30)
+		oldEnough := int64(15)
+		nonce := int64(0)
+		originalBroadcastAt := time.Unix(1616509100, 0)
+
+		etx := cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, borm, nonce, fromAddress, originalBroadcastAt)
+		attempt1 := etx.EthTxAttempts[0]
+		require.NoError(t, db.Get(&attempt1, `UPDATE eth_tx_attempts SET broadcast_before_block_num=$1 WHERE id=$2 RETURNING *`, oldEnough, attempt1.ID))
+		var err error
+
+		// Send transaction and assume success.
+		ethClient.On("SendTransaction", mock.Anything, mock.Anything).Return(nil).Once()
+
+		err = ec.RebroadcastWhereNecessary(testutils.Context(t), currentHead)
+		require.NoError(t, err)
+
+		etx, err = borm.FindEthTxWithAttempts(etx.ID)
+		require.NoError(t, err)
+		require.Len(t, etx.EthTxAttempts, 1)
+	})
+
+	t.Run("should retry previous attempt if connectivity check failed for dynamic transactions", func(t *testing.T) {
+		cfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+			c.EVM[0].GasEstimator.EIP1559DynamicFees = ptr(true)
+			c.EVM[0].GasEstimator.BlockHistory.BlockHistorySize = ptr[uint16](2)
+			c.EVM[0].GasEstimator.BlockHistory.CheckInclusionBlocks = ptr[uint16](4)
+		})
+		evmcfg := evmtest.NewChainScopedConfig(t, cfg)
+
+		borm := cltest.NewTxmORM(t, db, cfg)
+		ethKeyStore := cltest.NewKeyStore(t, db, cfg).Eth()
+		state, fromAddress := cltest.MustInsertRandomKeyReturningState(t, ethKeyStore)
+		keys := []ethkey.State{state}
+		kst := ksmocks.NewEth(t)
+
+		estimator := gasmocks.NewEvmEstimator(t)
+		estimator.On("BumpDynamicFee", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(gas.DynamicFee{}, uint32(0), pkgerrors.Wrapf(gas.ErrConnectivity, "transaction..."))
+		// Create confirmer with necessary state
+		ec := txmgr.NewEthConfirmer(borm, ethClient, evmcfg, kst, keys, gas.NewWrappedEvmEstimator(estimator, evmcfg), nil, lggr)
+		currentHead := int64(30)
+		oldEnough := int64(15)
+		nonce := int64(0)
+		originalBroadcastAt := time.Unix(1616509100, 0)
+
+		etx := cltest.MustInsertUnconfirmedEthTxWithBroadcastDynamicFeeAttempt(t, borm, nonce, fromAddress, originalBroadcastAt)
+		attempt1 := etx.EthTxAttempts[0]
+		require.NoError(t, db.Get(&attempt1, `UPDATE eth_tx_attempts SET broadcast_before_block_num=$1 WHERE id=$2 RETURNING *`, oldEnough, attempt1.ID))
+		var err error
+
+		// Send transaction and assume success.
+		ethClient.On("SendTransaction", mock.Anything, mock.Anything).Return(nil).Once()
+
+		err = ec.RebroadcastWhereNecessary(testutils.Context(t), currentHead)
+		require.NoError(t, err)
+
+		etx, err = borm.FindEthTxWithAttempts(etx.ID)
+		require.NoError(t, err)
+		require.Len(t, etx.EthTxAttempts, 1)
+	})
+}
+
 func TestEthConfirmer_RebroadcastWhereNecessary(t *testing.T) {
 	t.Parallel()
 

core/chains/evm/txmgr/models.go

@@ -302,6 +302,10 @@ type EthTxAttempt struct {
 	TxType                  int
 }
 
+func (a EthTxAttempt) String() string {
+	return fmt.Sprintf("EthTxAttempt(ID:%d,EthTxID:%d,GasPrice:%v,GasTipCap:%v,GasFeeCap:%v,TxType:%d", a.ID, a.EthTxID, a.GasPrice, a.GasTipCap, a.GasFeeCap, a.TxType)
+}
+
 // GetSignedTx decodes the SignedRawTx into a types.Transaction struct
 func (a EthTxAttempt) GetSignedTx() (*types.Transaction, error) {
 	s := rlp.NewStream(bytes.NewReader(a.SignedRawTx), 0)