Added confirmer test

common/txmgr/confirmer.go

@@ -469,22 +469,21 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Pro
 			purgeAttempt, err := ec.TxAttemptBuilder.NewPurgeTxAttempt(ctx, tx, lggr)
 			if err != nil {
 				errMu.Lock()
-				errorList = append(errorList, err)
+				errorList = append(errorList, fmt.Errorf("failed to create a purge attempt: %w", err))
 				errMu.Unlock()
 				return
 			}
 			// Save purge attempt
 			if err := ec.txStore.SaveInProgressAttempt(ctx, &purgeAttempt); err != nil {
 				errMu.Lock()
-				errorList = append(errorList, fmt.Errorf("saveInProgressAttempt failed for purge attempt: %w", err))
+				errorList = append(errorList, fmt.Errorf("failed to save purge attempt: %w", err))
 				errMu.Unlock()
 				return
 			}
-
 			// Send purge attempt
 			if err := ec.handleInProgressAttempt(ctx, lggr, tx, purgeAttempt, blockNum); err != nil {
 				errMu.Lock()
-				errorList = append(errorList, fmt.Errorf("handleInProgressAttempt failed for purge attempt: %w", err))
+				errorList = append(errorList, fmt.Errorf("failed to send purge attempt: %w", err))
 				errMu.Unlock()
 				return
 			}

core/chains/evm/txmgr/attempts.go

@@ -96,7 +96,6 @@ func (c *evmTxAttemptBuilder) NewPurgeTxAttempt(ctx context.Context, etx Tx, lgg
 	// Transactions being purged will always have a previous attempt since it had to have been broadcasted before at least once
 	previousAttempt := etx.TxAttempts[0]
 	keySpecificMaxGasPriceWei := c.feeConfig.PriceMaxKey(etx.FromAddress)
-	// TODO: Confirm if bump is needed. Either we could have a separate config or a hardcoded 10% to minimize the replacement fee
 	bumpedFee, _, err := c.EvmFeeEstimator.BumpFee(ctx, previousAttempt.TxFee, etx.FeeLimit, keySpecificMaxGasPriceWei, newEvmPriorAttempts(etx.TxAttempts))
 	if err != nil {
 		return attempt, fmt.Errorf("failed to bump previous fee to use for the purge attempt: %w", err)

core/chains/evm/txmgr/confirmer_test.go

@@ -3128,6 +3128,92 @@ func TestEthConfirmer_ResumePendingRuns(t *testing.T) {
 	})
 }
 
+func TestEthConfirmer_ProcessStuckTransactions(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	txStore := cltest.NewTestTxStore(t, db)
+	ethKeyStore := cltest.NewKeyStore(t, db).Eth()
+	_, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore)
+	ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+	ethClient.On("SendTransactionReturnCode", mock.Anything, mock.Anything, fromAddress).Return(commonclient.Successful, nil).Once()
+	lggr := logger.Test(t)
+	feeEstimator := gasmocks.NewEvmFeeEstimator(t)
+
+	// Return 10 gwei as market gas price
+	marketGasPrice := tenGwei
+	fee := gas.EvmFee{Legacy: marketGasPrice}
+	bumpedLegacy := assets.GWei(30)
+	bumpedFee := gas.EvmFee{Legacy: bumpedLegacy}
+	feeEstimator.On("GetFee", mock.Anything, []byte{}, uint64(0), mock.Anything).Return(fee, uint64(0), nil)
+	feeEstimator.On("BumpFee", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(bumpedFee, uint64(10_000), nil)
+	autoPurgeThreshold := uint32(5)
+	autoPurgeMinAttempts := uint32(3)
+	limitDefault := uint64(100)
+	cfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+		c.EVM[0].GasEstimator.LimitDefault = ptr(limitDefault)
+		c.EVM[0].Transactions.AutoPurge.AutoPurgeStuckTxs = ptr(true)
+		c.EVM[0].Transactions.AutoPurge.AutoPurgeThreshold = ptr(autoPurgeThreshold)
+		c.EVM[0].Transactions.AutoPurge.AutoPurgeMinAttempts = ptr(autoPurgeMinAttempts)
+	})
+	evmcfg := evmtest.NewChainScopedConfig(t, cfg)
+	ge := evmcfg.EVM().GasEstimator()
+	txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), ge, ethKeyStore, feeEstimator)
+	stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", evmcfg.EVM().Transactions().AutoPurge(), feeEstimator, txStore, ethClient)
+	ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), txmgr.NewEvmTxmConfig(evmcfg.EVM()), txmgr.NewEvmTxmFeeConfig(ge), evmcfg.EVM().Transactions(), cfg.Database(), ethKeyStore, txBuilder, lggr, stuckTxDetector)
+	servicetest.Run(t, ec)
+
+	ctx := testutils.Context(t)
+	blockNum := int64(100)
+
+	t.Run("detects and processes stuck transactions", func(t *testing.T) {
+		nonce := int64(0)
+		// Create attempts so that the oldest broadcast attempt's block num is what meets the threshold check
+		// Create autoPurgeMinAttempts number of attempts to ensure the broadcast attempt count check is not being triggered
+		// Create attempts broadcasted autoPurgeThreshold block ago to ensure broadcast block num check is not being triggered
+		tx := mustInsertUnconfirmedTxWithXBroadcastAttempts(t, txStore, nonce, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei))
+
+		// ProcessStuckTransactions should:
+		// 1. Detect a stuck transaction, confirmed above
+		// 2. Create a purge attempt for it
+		// 3. Save the purge attempt to the DB
+		// 4. Send the purge attempt
+		err := ec.ProcessStuckTransactions(ctx, blockNum)
+		require.NoError(t, err)
+
+		// Check if the purge attempt was saved to the DB properly
+		dbTx, err := txStore.FindTxWithAttempts(ctx, tx.ID)
+		require.NoError(t, err)
+		require.NotNil(t, dbTx)
+		latestAttempt := dbTx.TxAttempts[0]
+		require.Equal(t, true, latestAttempt.IsPurgeAttempt)
+		require.Equal(t, limitDefault, latestAttempt.ChainSpecificFeeLimit)
+		require.Equal(t, bumpedFee.Legacy, latestAttempt.TxFee.Legacy)
+
+		ethClient.On("SequenceAt", mock.Anything, mock.Anything, mock.Anything).Return(evmtypes.Nonce(1), nil)
+		ethClient.On("BatchCallContext", mock.Anything, mock.MatchedBy(func(b []rpc.BatchElem) bool {
+			return len(b) == 4 && cltest.BatchElemMatchesParams(b[0], latestAttempt.Hash, "eth_getTransactionReceipt")
+		})).Return(nil).Run(func(args mock.Arguments) {
+			elems := args.Get(1).([]rpc.BatchElem)
+			// First transaction confirmed
+			*(elems[0].Result.(*evmtypes.Receipt)) = evmtypes.Receipt{
+				TxHash:           latestAttempt.Hash,
+				BlockHash:        utils.NewHash(),
+				BlockNumber:      big.NewInt(101),
+				TransactionIndex: uint(1),
+				Status:           uint64(1),
+			}
+		}).Once()
+		// Check if transaction marked as fatal after purge attempt is confirmed
+		err = ec.CheckForReceipts(ctx, blockNum + 1)
+		require.NoError(t, err)
+		dbTx, err = txStore.FindTxWithAttempts(ctx, tx.ID)
+		require.NoError(t, err)
+		require.NotNil(t, dbTx)
+		require.Equal(t, txmgrcommon.TxFatalError, dbTx.State)
+	})
+}
+
 func ptr[T any](t T) *T { return &t }
 
 func newEthConfirmer(t testing.TB, txStore txmgr.EvmTxStore, ethClient client.Client, gconfig chainlink.GeneralConfig, config evmconfig.ChainScopedConfig, ks keystore.Eth, fn txmgrcommon.ResumeCallback) *txmgr.Confirmer {

core/chains/evm/txmgr/stuck_tx_detector.go

@@ -56,7 +56,9 @@ type stuckTxDetector struct {
 }
 
 func NewStuckTxDetector(lggr logger.Logger, chainID *big.Int, chainType config.ChainType, cfg stuckTxDetectorConfig, gasEstimator stuckTxDetectorGasEstimator, txStore stuckTxDetectorTxStore, chainClient stuckTxDetectorClient) *stuckTxDetector {
-	// TODO: ensure to initialize client with the usual security standards
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	t.DisableCompression = true
+	httpClient := &http.Client{Transport: t}
 	// TODO: Load purgeBlockNumMap with some DB state or confirm rate limit is not needed on first purge after restart
 	return &stuckTxDetector{
 		lggr:             lggr,
@@ -66,7 +68,7 @@ func NewStuckTxDetector(lggr logger.Logger, chainID *big.Int, chainType config.C
 		gasEstimator:     gasEstimator,
 		txStore:          txStore,
 		chainClient:      chainClient,
-		httpClient:       &http.Client{},
+		httpClient:       httpClient,
 		purgeBlockNumMap: make(map[common.Address]int64),
 	}
 }