Added purge block num loading on Confirmer startup

common/txmgr/confirmer.go

@@ -209,6 +209,10 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) sta
 	if err != nil {
 		return fmt.Errorf("Confirmer: failed to load EnabledAddressesForChain: %w", err)
 	}
+	err = ec.stuckTxDetector.LoadPurgeBlockNumMap(ctx, ec.enabledAddresses)
+	if err != nil {
+		ec.lggr.Debugf("Confirmer: failed to load the last purged block num for enabled addresses. Process can continue as normal but purge rate limiting may be affected.")
+	}
 
 	ec.ctx, ec.ctxCancel = context.WithCancel(context.Background())
 	ec.wg = sync.WaitGroup{}

common/txmgr/types/stuck_tx_detector.go

@@ -15,6 +15,7 @@ type StuckTxDetector[
 	FEE feetypes.Fee, // FEE - chain fee type
 ] interface {
 	DetectStuckTransactions(ctx context.Context, enabledAddresses []ADDR, blockNum int64) ([]Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error)
+	LoadPurgeBlockNumMap(ctx context.Context, addresses []ADDR) error
 	SetPurgeBlockNum(fromAddress ADDR, blockNum int64)
 	StuckTxFatalError() *string
 }

core/chains/evm/txmgr/confirmer_test.go

@@ -3171,7 +3171,7 @@ func TestEthConfirmer_ProcessStuckTransactions(t *testing.T) {
 		// 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))
+		tx := mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, nonce, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei))
 
 		head := evmtypes.Head{
 			Hash:   utils.NewHash(),

core/chains/evm/txmgr/evm_tx_store.go

@@ -56,7 +56,7 @@ type TxStoreWebApi interface {
 	TransactionsWithAttempts(ctx context.Context, offset, limit int) ([]Tx, int, error)
 	FindTxAttempt(ctx context.Context, hash common.Hash) (*TxAttempt, error)
 	FindTxWithAttempts(ctx context.Context, etxID int64) (etx Tx, err error)
-	FindUnconfirmedTxsByFromAddresses(ctx context.Context, addresses []common.Address, chainID *big.Int) (txs []Tx, err error)
+	FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state txmgrtypes.TxState, chainID *big.Int) (txs []*Tx, err error)
 }
 
 type TestEvmTxStore interface {
@@ -828,30 +828,34 @@ ORDER BY evm.txes.nonce ASC, evm.tx_attempts.gas_price DESC, evm.tx_attempts.gas
 	return
 }
 
-// Returns the latest attempt for the lowest nonce unconfirmed transaction in the DB if it is not marked as insufficient eth or for purge already
-func (o *evmTxStore) FindUnconfirmedTxsByFromAddresses(ctx context.Context, addresses []common.Address, chainID *big.Int) (txs []Tx, err error) {
+// Returns the transaction by state and from addresses
+// Loads attempt and receipts in the transactions
+func (o *evmTxStore) FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state txmgrtypes.TxState, chainID *big.Int) (txs []*Tx, err error) {
 	var cancel context.CancelFunc
 	ctx, cancel = o.mergeContexts(ctx)
 	defer cancel()
 	enabledAddrsBytea := make([][]byte, len(addresses))
 	for i, addr := range addresses {
 		enabledAddrsBytea[i] = addr.Bytes()
 	}
-	err = o.Transaction(ctx, true, func(orm *evmTxStore) error {
-		var dbTxs []DbEthTx
-		err = orm.q.SelectContext(ctx, &dbTxs, `
-SELECT evm.txes.* FROM evm.txes
-WHERE evm.txes.state = 'unconfirmed' AND evm.txes.from_address = ANY($1) AND evm.txes.evm_chain_id = $2
-`, enabledAddrsBytea, chainID.String())
+	err = o.Transact(ctx, true, func(orm *evmTxStore) error {
+		var dbEtxs []DbEthTx
+		err = orm.q.SelectContext(ctx, &dbEtxs, `SELECT * FROM evm.txes WHERE state = $1 AND from_address = ANY($2) AND evm_chain_id = $3`, state, enabledAddrsBytea, chainID.String())
 		if err != nil {
-			return fmt.Errorf("FindTxAttemptsForPurge failed to load evm.tx_attempts: %w", err)
+			return fmt.Errorf("FindTxsByStateAndFromAddresses failed to load evm.txes: %w", err)
 		}
-		if len(dbTxs) == 0 {
+		if len(dbEtxs) == 0 {
 			return nil
 		}
-		txs = dbEthTxsToEvmEthTxs(dbTxs)
-		err = orm.preloadTxAttempts(ctx, txs)
-		return pkgerrors.Wrap(err, "FindTxAttemptsForPurge failed to load evm.txes")
+		txs = make([]*Tx, len(dbEtxs))
+		dbEthTxsToEvmEthTxPtrs(dbEtxs, txs)
+		if err = orm.LoadTxesAttempts(ctx, txs); err != nil {
+			return fmt.Errorf("FindTxsByStateAndFromAddresses failed to load evm.tx_attempts: %w", err)
+		}
+		if err = orm.loadEthTxesAttemptsReceipts(ctx, txs); err != nil {
+			return fmt.Errorf("FindTxsByStateAndFromAddresses failed to load evm.receipts: %w", err)
+		}
+		return nil
 	})
 	return
 }

core/chains/evm/txmgr/stuck_tx_detector.go

@@ -16,6 +16,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink/v2/common/config"
 	feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types"
+	"github.com/smartcontractkit/chainlink/v2/common/txmgr"
 	"github.com/smartcontractkit/chainlink/v2/common/txmgr/types"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/gas"
@@ -30,7 +31,7 @@ type stuckTxDetectorClient interface {
 }
 
 type stuckTxDetectorTxStore interface {
-	FindUnconfirmedTxsByFromAddresses(ctx context.Context, addresses []common.Address, chainID *big.Int) (txs []Tx, err error)
+	FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state types.TxState, chainID *big.Int) (txs []*Tx, err error)
 }
 
 type stuckTxDetectorConfig interface {
@@ -59,7 +60,6 @@ func NewStuckTxDetector(lggr logger.Logger, chainID *big.Int, chainType config.C
 	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,
 		chainID:          chainID,
@@ -73,6 +73,35 @@ func NewStuckTxDetector(lggr logger.Logger, chainID *big.Int, chainType config.C
 	}
 }
 
+func (d *stuckTxDetector) LoadPurgeBlockNumMap(ctx context.Context, addresses []common.Address) error {
+	d.purgeBlockNumLock.Lock()
+	defer d.purgeBlockNumLock.Unlock()
+	// Ok to reset the map here since this method could be reloaded with a new list of from addresses
+	d.purgeBlockNumMap = make(map[common.Address]int64)
+	for _, address := range addresses {
+		d.purgeBlockNumMap[address] = 0
+	}
+
+	// Find all fatal error transactions to see if any were from previous purges to properly set the map
+	txs, err := d.txStore.FindTxsByStateAndFromAddresses(ctx, addresses, txmgr.TxFatalError, d.chainID)
+	if err != nil {
+		return fmt.Errorf("failed to query fatal error transactions from the txstore: %w", err)
+	}
+
+	// Set the purgeBlockNumMap with the receipt block num of purge attempts
+	for _, tx := range txs {
+		for _, attempt := range tx.TxAttempts {
+			if attempt.IsPurgeAttempt && len(attempt.Receipts) > 0 {
+				// There should only be 1 receipt in an attempt for a transaction
+				d.purgeBlockNumMap[tx.FromAddress] = attempt.Receipts[0].GetBlockNumber().Int64()
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
 // If the AutoPurgeStuckTxs feature is enabled, finds terminally stuck transactions
 // Uses a chain specific method for detection, or if one does not exist, applies a general heuristic
 func (d *stuckTxDetector) DetectStuckTransactions(ctx context.Context, enabledAddresses []common.Address, blockNum int64) ([]Tx, error) {
@@ -100,7 +129,7 @@ func (d *stuckTxDetector) DetectStuckTransactions(ctx context.Context, enabledAd
 
 	for _, stuckTx := range stuckTxs {
 		lggr := stuckTx.GetLogger(d.lggr)
-		lggr.Errorw("marking transaction as terminally stuck", "etx", stuckTx)
+		lggr.Debugw("marking transaction as terminally stuck", "etx", stuckTx)
 	}
 
 	return stuckTxs, err
@@ -110,17 +139,17 @@ func (d *stuckTxDetector) DetectStuckTransactions(ctx context.Context, enabledAd
 // Only the earliest transaction can be considered terminally stuck. All others may be valid and just stuck behind the nonce
 func (d *stuckTxDetector) FindPotentialStuckTxs(ctx context.Context, enabledAddresses []common.Address) ([]Tx, error) {
 	// Loads attempts within tx
-	txs, err := d.txStore.FindUnconfirmedTxsByFromAddresses(ctx, enabledAddresses, d.chainID)
+	txs, err := d.txStore.FindTxsByStateAndFromAddresses(ctx, enabledAddresses, txmgr.TxUnconfirmed, d.chainID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve unconfirmed transactions for enabled addresses: %w", err)
 	}
 	// Stores the lowest nonce tx found in the query results for each from address
 	lowestNonceTxMap := make(map[common.Address]Tx)
 	for _, tx := range txs {
 		if _, ok := lowestNonceTxMap[tx.FromAddress]; !ok {
-			lowestNonceTxMap[tx.FromAddress] = tx
+			lowestNonceTxMap[tx.FromAddress] = *tx
 		} else if lowestNonceTx := lowestNonceTxMap[tx.FromAddress]; *lowestNonceTx.Sequence > *tx.Sequence {
-			lowestNonceTxMap[tx.FromAddress] = tx
+			lowestNonceTxMap[tx.FromAddress] = *tx
 		}
 	}