diff --git a/.changeset/red-eagles-cry.md b/.changeset/red-eagles-cry.md new file mode 100644 index 00000000000..3474435911d --- /dev/null +++ b/.changeset/red-eagles-cry.md @@ -0,0 +1,5 @@ +--- +"chainlink": minor +--- + +#added Added an auto-purge feature to the EVM TXM that identifies terminally stuck transactions either through a chain specific method or heurisitic then purges them to unblock the nonce. Included 4 new toml configs under Transactions.AutoPurge to configure this new feature: Enabled, Threshold, MinAttempts, and DetectionApiUrl. diff --git a/common/config/chaintype.go b/common/config/chaintype.go index 3f3150950d6..1ddb3f626b5 100644 --- a/common/config/chaintype.go +++ b/common/config/chaintype.go @@ -17,6 +17,7 @@ const ( ChainScroll ChainType = "scroll" ChainWeMix ChainType = "wemix" ChainXLayer ChainType = "xlayer" + ChainZkEvm ChainType = "zkevm" ChainZkSync ChainType = "zksync" ) @@ -34,7 +35,7 @@ func (c ChainType) IsL2() bool { func (c ChainType) IsValid() bool { switch c { - case "", ChainArbitrum, ChainCelo, ChainGnosis, ChainKroma, ChainMetis, ChainOptimismBedrock, ChainScroll, ChainWeMix, ChainXLayer, ChainZkSync: + case "", ChainArbitrum, ChainCelo, ChainGnosis, ChainKroma, ChainMetis, ChainOptimismBedrock, ChainScroll, ChainWeMix, ChainXLayer, ChainZkEvm, ChainZkSync: return true } return false @@ -60,6 +61,8 @@ func ChainTypeFromSlug(slug string) ChainType { return ChainWeMix case "xlayer": return ChainXLayer + case "zkevm": + return ChainZkEvm case "zksync": return ChainZkSync default: @@ -123,5 +126,6 @@ var ErrInvalidChainType = fmt.Errorf("must be one of %s or omitted", strings.Joi string(ChainScroll), string(ChainWeMix), string(ChainXLayer), + string(ChainZkEvm), string(ChainZkSync), }, ", ")) diff --git a/common/txmgr/confirmer.go b/common/txmgr/confirmer.go index 30fbbc48987..294e922c1c0 100644 --- a/common/txmgr/confirmer.go +++ b/common/txmgr/confirmer.go @@ -122,12 +122,13 @@ type Confirmer[ lggr logger.SugaredLogger client txmgrtypes.TxmClient[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] txmgrtypes.TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] - resumeCallback ResumeCallback - chainConfig txmgrtypes.ConfirmerChainConfig - feeConfig txmgrtypes.ConfirmerFeeConfig - txConfig txmgrtypes.ConfirmerTransactionsConfig - dbConfig txmgrtypes.ConfirmerDatabaseConfig - chainID CHAIN_ID + stuckTxDetector txmgrtypes.StuckTxDetector[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] + resumeCallback ResumeCallback + chainConfig txmgrtypes.ConfirmerChainConfig + feeConfig txmgrtypes.ConfirmerFeeConfig + txConfig txmgrtypes.ConfirmerTransactionsConfig + dbConfig txmgrtypes.ConfirmerDatabaseConfig + chainID CHAIN_ID ks txmgrtypes.KeyStore[ADDR, CHAIN_ID, SEQ] enabledAddresses []ADDR @@ -162,6 +163,7 @@ func NewConfirmer[ txAttemptBuilder txmgrtypes.TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], lggr logger.Logger, isReceiptNil func(R) bool, + stuckTxDetector txmgrtypes.StuckTxDetector[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], ) *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] { lggr = logger.Named(lggr, "Confirmer") return &Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{ @@ -178,6 +180,7 @@ func NewConfirmer[ ks: keystore, mb: mailbox.NewSingle[HEAD](), isReceiptNil: isReceiptNil, + stuckTxDetector: stuckTxDetector, } } @@ -205,6 +208,9 @@ 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) } + if err = ec.stuckTxDetector.LoadPurgeBlockNumMap(ctx, ec.enabledAddresses); 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.stopCh = make(chan struct{}) ec.wg = sync.WaitGroup{} @@ -298,6 +304,13 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) pro ec.lggr.Debugw("Finished CheckForReceipts", "headNum", head.BlockNumber(), "time", time.Since(mark), "id", "confirmer") mark = time.Now() + if err := ec.ProcessStuckTransactions(ctx, head.BlockNumber()); err != nil { + return fmt.Errorf("ProcessStuckTransactions failed: %w", err) + } + + ec.lggr.Debugw("Finished ProcessStuckTransactions", "headNum", head.BlockNumber(), "time", time.Since(mark), "id", "confirmer") + mark = time.Now() + if err := ec.RebroadcastWhereNecessary(ctx, head.BlockNumber()); err != nil { return fmt.Errorf("RebroadcastWhereNecessary failed: %w", err) } @@ -436,6 +449,57 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che return nil } +// Determines if any of the unconfirmed transactions are terminally stuck for each enabled address +// If any transaction is found to be terminally stuck, this method sends an empty attempt with bumped gas in an attempt to purge the stuck transaction +func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) ProcessStuckTransactions(ctx context.Context, blockNum int64) error { + // Use the detector to find a stuck tx for each enabled address + stuckTxs, err := ec.stuckTxDetector.DetectStuckTransactions(ctx, ec.enabledAddresses, blockNum) + if err != nil { + return fmt.Errorf("failed to detect stuck transactions: %w", err) + } + if len(stuckTxs) == 0 { + return nil + } + + var wg sync.WaitGroup + wg.Add(len(stuckTxs)) + errorList := []error{} + var errMu sync.Mutex + for _, tx := range stuckTxs { + // All stuck transactions will have unique from addresses. It is safe to process separate keys concurrently + // NOTE: This design will block one key if another takes a really long time to execute + go func(tx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) { + defer wg.Done() + lggr := tx.GetLogger(ec.lggr) + // Create an purge attempt for tx + purgeAttempt, err := ec.TxAttemptBuilder.NewPurgeTxAttempt(ctx, tx, lggr) + if err != nil { + errMu.Lock() + 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("failed to save purge attempt: %w", err)) + errMu.Unlock() + return + } + lggr.Warnw("marked transaction as terminally stuck", "etx", tx) + // Send purge attempt + if err := ec.handleInProgressAttempt(ctx, lggr, tx, purgeAttempt, blockNum); err != nil { + errMu.Lock() + errorList = append(errorList, fmt.Errorf("failed to send purge attempt: %w", err)) + errMu.Unlock() + return + } + }(tx) + } + wg.Wait() + return errors.Join(errorList...) +} + func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) separateLikelyConfirmedAttempts(from ADDR, attempts []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], minedSequence SEQ) []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] { if len(attempts) == 0 { return attempts @@ -486,7 +550,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) fet j = len(attempts) } - ec.lggr.Debugw(fmt.Sprintf("Batch fetching receipts at indexes %v until (excluded) %v", i, j), "blockNum", blockNum) + ec.lggr.Debugw(fmt.Sprintf("Batch fetching receipts at indexes %d until (excluded) %d", i, j), "blockNum", blockNum) batch := attempts[i:j] @@ -494,7 +558,13 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) fet if err != nil { return fmt.Errorf("batchFetchReceipts failed: %w", err) } - if err := ec.txStore.SaveFetchedReceipts(ctx, receipts, ec.chainID); err != nil { + validReceipts, purgeReceipts := ec.separateValidAndPurgeAttemptReceipts(receipts, batch) + // Saves the receipts and mark the associated transactions as Confirmed + if err := ec.txStore.SaveFetchedReceipts(ctx, validReceipts, TxConfirmed, nil, ec.chainID); err != nil { + return fmt.Errorf("saveFetchedReceipts failed: %w", err) + } + // Save the receipts but mark the associated transactions as Fatal Error since the original transaction was purged + if err := ec.txStore.SaveFetchedReceipts(ctx, purgeReceipts, TxFatalError, ec.stuckTxDetector.StuckTxFatalError(), ec.chainID); err != nil { return fmt.Errorf("saveFetchedReceipts failed: %w", err) } promNumConfirmedTxs.WithLabelValues(ec.chainID.String()).Add(float64(len(receipts))) @@ -507,6 +577,25 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) fet return nil } +func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) separateValidAndPurgeAttemptReceipts(receipts []R, attempts []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) (valid []R, purge []R) { + receiptMap := make(map[TX_HASH]R) + for _, receipt := range receipts { + receiptMap[receipt.GetTxHash()] = receipt + } + for _, attempt := range attempts { + if receipt, ok := receiptMap[attempt.Hash]; ok { + if attempt.IsPurgeAttempt { + // Setting the purged block num here is ok since we have confirmation the tx has been purged with the receipt + ec.stuckTxDetector.SetPurgeBlockNum(attempt.Tx.FromAddress, receipt.GetBlockNumber().Int64()) + purge = append(purge, receipt) + } else { + valid = append(valid, receipt) + } + } + } + return +} + func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) getMinedSequenceForAddress(ctx context.Context, from ADDR) (SEQ, error) { return ec.client.SequenceAt(ctx, from, nil) } @@ -700,7 +789,6 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Fin lggr.Infow(fmt.Sprintf("Found %d transactions to be re-sent that were previously rejected due to insufficient native token balance", len(etxInsufficientFunds)), "blockNum", blockNum, "address", address) } - // TODO: Just pass the Q through everything etxBumps, err := ec.txStore.FindTxsRequiringGasBump(ctx, address, blockNum, gasBumpThreshold, bumpDepth, chainID) if ctx.Err() != nil { return nil, nil diff --git a/common/txmgr/types/mocks/tx_attempt_builder.go b/common/txmgr/types/mocks/tx_attempt_builder.go index 8171b29bbed..47614674336 100644 --- a/common/txmgr/types/mocks/tx_attempt_builder.go +++ b/common/txmgr/types/mocks/tx_attempt_builder.go @@ -188,6 +188,34 @@ func (_m *TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) return r0, r1 } +// NewPurgeTxAttempt provides a mock function with given fields: ctx, etx, lggr +func (_m *TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) NewPurgeTxAttempt(ctx context.Context, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], lggr logger.Logger) (txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error) { + ret := _m.Called(ctx, etx, lggr) + + if len(ret) == 0 { + panic("no return value specified for NewPurgeTxAttempt") + } + + var r0 txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], logger.Logger) (txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error)); ok { + return rf(ctx, etx, lggr) + } + if rf, ok := ret.Get(0).(func(context.Context, txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], logger.Logger) txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]); ok { + r0 = rf(ctx, etx, lggr) + } else { + r0 = ret.Get(0).(txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) + } + + if rf, ok := ret.Get(1).(func(context.Context, txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], logger.Logger) error); ok { + r1 = rf(ctx, etx, lggr) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // NewTxAttempt provides a mock function with given fields: ctx, tx, lggr, opts func (_m *TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) NewTxAttempt(ctx context.Context, tx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], lggr logger.Logger, opts ...feetypes.Opt) (txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], FEE, uint64, bool, error) { _va := make([]interface{}, len(opts)) diff --git a/common/txmgr/types/mocks/tx_store.go b/common/txmgr/types/mocks/tx_store.go index be2c0aef723..bb8fabf1a71 100644 --- a/common/txmgr/types/mocks/tx_store.go +++ b/common/txmgr/types/mocks/tx_store.go @@ -1014,17 +1014,17 @@ func (_m *TxStore[ADDR, CHAIN_ID, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) SaveConfirm return r0 } -// SaveFetchedReceipts provides a mock function with given fields: ctx, receipts, chainID -func (_m *TxStore[ADDR, CHAIN_ID, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) SaveFetchedReceipts(ctx context.Context, receipts []R, chainID CHAIN_ID) error { - ret := _m.Called(ctx, receipts, chainID) +// SaveFetchedReceipts provides a mock function with given fields: ctx, r, state, errorMsg, chainID +func (_m *TxStore[ADDR, CHAIN_ID, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) SaveFetchedReceipts(ctx context.Context, r []R, state txmgrtypes.TxState, errorMsg *string, chainID CHAIN_ID) error { + ret := _m.Called(ctx, r, state, errorMsg, chainID) if len(ret) == 0 { panic("no return value specified for SaveFetchedReceipts") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, []R, CHAIN_ID) error); ok { - r0 = rf(ctx, receipts, chainID) + if rf, ok := ret.Get(0).(func(context.Context, []R, txmgrtypes.TxState, *string, CHAIN_ID) error); ok { + r0 = rf(ctx, r, state, errorMsg, chainID) } else { r0 = ret.Error(0) } diff --git a/common/txmgr/types/stuck_tx_detector.go b/common/txmgr/types/stuck_tx_detector.go new file mode 100644 index 00000000000..c4ca94b87f8 --- /dev/null +++ b/common/txmgr/types/stuck_tx_detector.go @@ -0,0 +1,26 @@ +package types + +import ( + "context" + + feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types" + "github.com/smartcontractkit/chainlink/v2/common/types" +) + +// StuckTxDetector is used by the Confirmer to determine if any unconfirmed transactions are terminally stuck +type StuckTxDetector[ + CHAIN_ID types.ID, // CHAIN_ID - chain id type + ADDR types.Hashable, // ADDR - chain address type + TX_HASH, BLOCK_HASH types.Hashable, // various chain hash types + SEQ types.Sequence, // SEQ - chain sequence type (nonce, utxo, etc) + FEE feetypes.Fee, // FEE - chain fee type +] interface { + // Uses either a chain specific API or heuristic to determine if any unconfirmed transactions are terminally stuck. Returns only one transaction per enabled address. + DetectStuckTransactions(ctx context.Context, enabledAddresses []ADDR, blockNum int64) ([]Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error) + // Loads the internal map that tracks the last block num a transaction was purged at using the DB state + LoadPurgeBlockNumMap(ctx context.Context, addresses []ADDR) error + // Sets the last purged block num after a transaction has been successfully purged with receipt + SetPurgeBlockNum(fromAddress ADDR, blockNum int64) + // Returns the error message to set in the transaction error field to mark it as terminally stuck + StuckTxFatalError() *string +} diff --git a/common/txmgr/types/tx.go b/common/txmgr/types/tx.go index 76fd7f4b3ab..e04ebe95871 100644 --- a/common/txmgr/types/tx.go +++ b/common/txmgr/types/tx.go @@ -183,6 +183,7 @@ type TxAttempt[ State TxAttemptState Receipts []ChainReceipt[TX_HASH, BLOCK_HASH] `json:"-"` TxType int + IsPurgeAttempt bool } func (a *TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) String() string { diff --git a/common/txmgr/types/tx_attempt_builder.go b/common/txmgr/types/tx_attempt_builder.go index 54184733f0a..a9cbddfc2cf 100644 --- a/common/txmgr/types/tx_attempt_builder.go +++ b/common/txmgr/types/tx_attempt_builder.go @@ -42,4 +42,7 @@ type TxAttemptBuilder[ // NewEmptyTxAttempt is used in ForceRebroadcast to create a signed tx with zero value sent to the zero address NewEmptyTxAttempt(ctx context.Context, seq SEQ, feeLimit uint64, fee FEE, fromAddress ADDR) (attempt TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error) + + // NewPurgeTxAttempt is used to create empty transaction attempts with higher gas than the previous attempt to purge stuck transactions + NewPurgeTxAttempt(ctx context.Context, etx Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], lggr logger.Logger) (attempt TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error) } diff --git a/common/txmgr/types/tx_store.go b/common/txmgr/types/tx_store.go index bca2d1e3647..2b82e1f6483 100644 --- a/common/txmgr/types/tx_store.go +++ b/common/txmgr/types/tx_store.go @@ -39,7 +39,7 @@ type TxStore[ FindTxesPendingCallback(ctx context.Context, blockNum int64, chainID CHAIN_ID) (receiptsPlus []ReceiptPlus[R], err error) // Update tx to mark that its callback has been signaled UpdateTxCallbackCompleted(ctx context.Context, pipelineTaskRunRid uuid.UUID, chainId CHAIN_ID) error - SaveFetchedReceipts(ctx context.Context, receipts []R, chainID CHAIN_ID) (err error) + SaveFetchedReceipts(ctx context.Context, r []R, state TxState, errorMsg *string, chainID CHAIN_ID) error // additional methods for tx store management CheckTxQueueCapacity(ctx context.Context, fromAddress ADDR, maxQueuedTransactions uint64, chainID CHAIN_ID) (err error) diff --git a/core/chains/evm/config/chain_scoped_transactions.go b/core/chains/evm/config/chain_scoped_transactions.go index df2343d3158..87031a4c66e 100644 --- a/core/chains/evm/config/chain_scoped_transactions.go +++ b/core/chains/evm/config/chain_scoped_transactions.go @@ -1,6 +1,7 @@ package config import ( + "net/url" "time" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/config/toml" @@ -33,3 +34,27 @@ func (t *transactionsConfig) MaxInFlight() uint32 { func (t *transactionsConfig) MaxQueued() uint64 { return uint64(*t.c.MaxQueued) } + +func (t *transactionsConfig) AutoPurge() AutoPurgeConfig { + return &autoPurgeConfig{c: t.c.AutoPurge} +} + +type autoPurgeConfig struct { + c toml.AutoPurgeConfig +} + +func (a *autoPurgeConfig) Enabled() bool { + return *a.c.Enabled +} + +func (a *autoPurgeConfig) Threshold() uint32 { + return *a.c.Threshold +} + +func (a *autoPurgeConfig) MinAttempts() uint32 { + return *a.c.MinAttempts +} + +func (a *autoPurgeConfig) DetectionApiUrl() *url.URL { + return a.c.DetectionApiUrl.URL() +} diff --git a/core/chains/evm/config/config.go b/core/chains/evm/config/config.go index 34de754de21..18f63b8d918 100644 --- a/core/chains/evm/config/config.go +++ b/core/chains/evm/config/config.go @@ -2,6 +2,7 @@ package config import ( "math/big" + "net/url" "time" gethcommon "github.com/ethereum/go-ethereum/common" @@ -100,6 +101,14 @@ type Transactions interface { ReaperThreshold() time.Duration MaxInFlight() uint32 MaxQueued() uint64 + AutoPurge() AutoPurgeConfig +} + +type AutoPurgeConfig interface { + Enabled() bool + Threshold() uint32 + MinAttempts() uint32 + DetectionApiUrl() *url.URL } //go:generate mockery --quiet --name GasEstimator --output ./mocks/ --case=underscore diff --git a/core/chains/evm/config/toml/config.go b/core/chains/evm/config/toml/config.go index a326881bdde..a22f6e7f557 100644 --- a/core/chains/evm/config/toml/config.go +++ b/core/chains/evm/config/toml/config.go @@ -393,6 +393,44 @@ func (c *Chain) ValidateConfig() (err error) { Msg: "must be greater than or equal to 1"}) } + // AutoPurge configs depend on ChainType so handling validation on per chain basis + if c.Transactions.AutoPurge.Enabled != nil && *c.Transactions.AutoPurge.Enabled { + chainType := c.ChainType.ChainType() + switch chainType { + case config.ChainScroll: + if c.Transactions.AutoPurge.DetectionApiUrl == nil { + err = multierr.Append(err, commonconfig.ErrMissing{Name: "Transactions.AutoPurge.DetectionApiUrl", Msg: fmt.Sprintf("must be set for %s", chainType)}) + } else if c.Transactions.AutoPurge.DetectionApiUrl.IsZero() { + err = multierr.Append(err, commonconfig.ErrInvalid{Name: "Transactions.AutoPurge.DetectionApiUrl", Value: c.Transactions.AutoPurge.DetectionApiUrl, Msg: fmt.Sprintf("must be set for %s", chainType)}) + } else { + switch c.Transactions.AutoPurge.DetectionApiUrl.Scheme { + case "http", "https": + default: + err = multierr.Append(err, commonconfig.ErrInvalid{Name: "Transactions.AutoPurge.DetectionApiUrl", Value: c.Transactions.AutoPurge.DetectionApiUrl.Scheme, Msg: "must be http or https"}) + } + } + case config.ChainZkEvm: + // No other configs are needed + default: + // Bump Threshold is required because the stuck tx heuristic relies on a minimum number of bump attempts to exist + if c.GasEstimator.BumpThreshold == nil { + err = multierr.Append(err, commonconfig.ErrMissing{Name: "GasEstimator.BumpThreshold", Msg: fmt.Sprintf("must be set if auto-purge feature is enabled for %s", chainType)}) + } else if *c.GasEstimator.BumpThreshold == 0 { + err = multierr.Append(err, commonconfig.ErrInvalid{Name: "GasEstimator.BumpThreshold", Value: 0, Msg: fmt.Sprintf("cannot be 0 if auto-purge feature is enabled for %s", chainType)}) + } + if c.Transactions.AutoPurge.Threshold == nil { + err = multierr.Append(err, commonconfig.ErrMissing{Name: "Transactions.AutoPurge.Threshold", Msg: fmt.Sprintf("needs to be set if auto-purge feature is enabled for %s", chainType)}) + } else if *c.Transactions.AutoPurge.Threshold == 0 { + err = multierr.Append(err, commonconfig.ErrInvalid{Name: "Transactions.AutoPurge.Threshold", Value: 0, Msg: fmt.Sprintf("cannot be 0 if auto-purge feature is enabled for %s", chainType)}) + } + if c.Transactions.AutoPurge.MinAttempts == nil { + err = multierr.Append(err, commonconfig.ErrMissing{Name: "Transactions.AutoPurge.MinAttempts", Msg: fmt.Sprintf("needs to be set if auto-purge feature is enabled for %s", chainType)}) + } else if *c.Transactions.AutoPurge.MinAttempts == 0 { + err = multierr.Append(err, commonconfig.ErrInvalid{Name: "Transactions.AutoPurge.MinAttempts", Value: 0, Msg: fmt.Sprintf("cannot be 0 if auto-purge feature is enabled for %s", chainType)}) + } + } + } + return } @@ -403,6 +441,8 @@ type Transactions struct { ReaperInterval *commonconfig.Duration ReaperThreshold *commonconfig.Duration ResendAfterThreshold *commonconfig.Duration + + AutoPurge AutoPurgeConfig `toml:",omitempty"` } func (t *Transactions) setFrom(f *Transactions) { @@ -424,6 +464,29 @@ func (t *Transactions) setFrom(f *Transactions) { if v := f.ResendAfterThreshold; v != nil { t.ResendAfterThreshold = v } + t.AutoPurge.setFrom(&f.AutoPurge) +} + +type AutoPurgeConfig struct { + Enabled *bool + Threshold *uint32 + MinAttempts *uint32 + DetectionApiUrl *commonconfig.URL +} + +func (a *AutoPurgeConfig) setFrom(f *AutoPurgeConfig) { + if v := f.Enabled; v != nil { + a.Enabled = v + } + if v := f.Threshold; v != nil { + a.Threshold = v + } + if v := f.MinAttempts; v != nil { + a.MinAttempts = v + } + if v := f.DetectionApiUrl; v != nil { + a.DetectionApiUrl = v + } } type OCR2 struct { diff --git a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Cardona.toml b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Cardona.toml index 9fa2b78e086..cd91465dae6 100644 --- a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Cardona.toml +++ b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Cardona.toml @@ -1,4 +1,5 @@ ChainID = '2442' +ChainType = 'zkevm' FinalityDepth = 500 NoNewHeadsThreshold = '12m' MinIncomingConfirmations = 1 diff --git a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Goerli.toml b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Goerli.toml index a259e4766f8..6a9b47190fd 100644 --- a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Goerli.toml +++ b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Goerli.toml @@ -1,4 +1,5 @@ ChainID = '1442' +ChainType = 'zkevm' FinalityDepth = 500 NoNewHeadsThreshold = '12m' MinIncomingConfirmations = 1 diff --git a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Mainnet.toml b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Mainnet.toml index e8833bc7312..79e0cb0fce5 100644 --- a/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Mainnet.toml +++ b/core/chains/evm/config/toml/defaults/Polygon_Zkevm_Mainnet.toml @@ -1,4 +1,5 @@ ChainID = '1101' +ChainType = 'zkevm' FinalityDepth = 500 NoNewHeadsThreshold = '6m' MinIncomingConfirmations = 1 diff --git a/core/chains/evm/config/toml/defaults/fallback.toml b/core/chains/evm/config/toml/defaults/fallback.toml index d65d0a1b0c1..9d4ddf7bf0a 100644 --- a/core/chains/evm/config/toml/defaults/fallback.toml +++ b/core/chains/evm/config/toml/defaults/fallback.toml @@ -23,6 +23,9 @@ ReaperInterval = '1h' ReaperThreshold = '168h' ResendAfterThreshold = '1m' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true diff --git a/core/chains/evm/txmgr/attempts.go b/core/chains/evm/txmgr/attempts.go index bf2d9a68edf..8566adcb5c5 100644 --- a/core/chains/evm/txmgr/attempts.go +++ b/core/chains/evm/txmgr/attempts.go @@ -37,6 +37,7 @@ type evmTxAttemptBuilderFeeConfig interface { TipCapMin() *assets.Wei PriceMin() *assets.Wei PriceMaxKey(common.Address) *assets.Wei + LimitDefault() uint64 } func NewEvmTxAttemptBuilder(chainID big.Int, feeConfig evmTxAttemptBuilderFeeConfig, keystore TxAttemptSigner[common.Address], estimator gas.EvmFeeEstimator) *evmTxAttemptBuilder { @@ -75,11 +76,41 @@ func (c *evmTxAttemptBuilder) NewBumpTxAttempt(ctx context.Context, etx Tx, prev if err != nil { return attempt, bumpedFee, bumpedFeeLimit, true, pkgerrors.Wrap(err, "failed to bump fee") // estimator errors are retryable } - + // If transaction's previous attempt is marked for purge, ensure the new bumped attempt also sends empty payload, 0 value, and LimitDefault as fee limit + if previousAttempt.IsPurgeAttempt { + etx.EncodedPayload = []byte{} + etx.Value = *big.NewInt(0) + bumpedFeeLimit = c.feeConfig.LimitDefault() + } attempt, retryable, err = c.NewCustomTxAttempt(ctx, etx, bumpedFee, bumpedFeeLimit, previousAttempt.TxType, lggr) + // If transaction's previous attempt is marked for purge, ensure the new bumped attempt is also marked for purge + if previousAttempt.IsPurgeAttempt { + attempt.IsPurgeAttempt = true + } return attempt, bumpedFee, bumpedFeeLimit, retryable, err } +func (c *evmTxAttemptBuilder) NewPurgeTxAttempt(ctx context.Context, etx Tx, lggr logger.Logger) (attempt TxAttempt, err error) { + // Use the LimitDefault since this is an empty tx + gasLimit := c.feeConfig.LimitDefault() + // 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) + 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) + } + // Set empty payload and 0 value for purge attempts + etx.EncodedPayload = []byte{} + etx.Value = *big.NewInt(0) + attempt, _, err = c.NewCustomTxAttempt(ctx, etx, bumpedFee, gasLimit, previousAttempt.TxType, lggr) + if err != nil { + return attempt, fmt.Errorf("failed to create purge attempt: %w", err) + } + attempt.IsPurgeAttempt = true + return attempt, nil +} + // NewCustomTxAttempt is the lowest level func where the fee parameters + tx type must be passed in // used in the txm for force rebroadcast where fees and tx type are pre-determined without an estimator func (c *evmTxAttemptBuilder) NewCustomTxAttempt(ctx context.Context, etx Tx, fee gas.EvmFee, gasLimit uint64, txType int, lggr logger.Logger) (attempt TxAttempt, retryable bool, err error) { diff --git a/core/chains/evm/txmgr/attempts_test.go b/core/chains/evm/txmgr/attempts_test.go index d5c8f577ce1..52340ce51a5 100644 --- a/core/chains/evm/txmgr/attempts_test.go +++ b/core/chains/evm/txmgr/attempts_test.go @@ -36,6 +36,7 @@ type feeConfig struct { tipCapMin *assets.Wei priceMin *assets.Wei priceMax *assets.Wei + limitDefault uint64 } func newFeeConfig() *feeConfig { @@ -50,6 +51,7 @@ func (g *feeConfig) EIP1559DynamicFees() bool { return g. func (g *feeConfig) TipCapMin() *assets.Wei { return g.tipCapMin } func (g *feeConfig) PriceMin() *assets.Wei { return g.priceMin } func (g *feeConfig) PriceMaxKey(addr gethcommon.Address) *assets.Wei { return g.priceMax } +func (g *feeConfig) LimitDefault() uint64 { return g.limitDefault } func TestTxm_SignTx(t *testing.T) { t.Parallel() @@ -224,6 +226,87 @@ func TestTxm_NewLegacyAttempt(t *testing.T) { }) } +func TestTxm_NewPurgeAttempt(t *testing.T) { + addr := NewEvmAddress() + kst := ksmocks.NewEth(t) + tx := types.NewTx(&types.LegacyTx{}) + kst.On("SignTx", mock.Anything, addr, mock.Anything, big.NewInt(1)).Return(tx, nil) + gc := newFeeConfig() + gc.priceMin = assets.GWei(10) + gc.priceMax = assets.GWei(50) + gc.limitDefault = uint64(10) + est := gasmocks.NewEvmFeeEstimator(t) + bumpedLegacy := assets.GWei(30) + bumpedDynamicFee := assets.GWei(15) + bumpedDynamicTip := assets.GWei(10) + bumpedFee := gas.EvmFee{Legacy: bumpedLegacy, DynamicTipCap: bumpedDynamicTip, DynamicFeeCap: bumpedDynamicFee} + est.On("BumpFee", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(bumpedFee, uint64(10_000), nil) + cks := txmgr.NewEvmTxAttemptBuilder(*big.NewInt(1), gc, kst, est) + lggr := logger.Test(t) + ctx := testutils.Context(t) + + t.Run("creates legacy purge attempt with fields if previous attempt is legacy", func(t *testing.T) { + n := evmtypes.Nonce(0) + etx := txmgr.Tx{Sequence: &n, FromAddress: addr, EncodedPayload: []byte{1, 2, 3}} + prevAttempt, _, err := cks.NewCustomTxAttempt(ctx, etx, gas.EvmFee{Legacy: bumpedLegacy.Sub(assets.GWei(1))}, 100, 0x0, lggr) + require.NoError(t, err) + etx.TxAttempts = append(etx.TxAttempts, prevAttempt) + a, err := cks.NewPurgeTxAttempt(ctx, etx, lggr) + require.NoError(t, err) + // The fee limit is overridden with LimitDefault since purge attempts are just empty attempts + require.Equal(t, gc.limitDefault, a.ChainSpecificFeeLimit) + require.NotNil(t, a.TxFee.Legacy) + require.Equal(t, bumpedLegacy.String(), a.TxFee.Legacy.String()) + require.Nil(t, a.TxFee.DynamicTipCap) + require.Nil(t, a.TxFee.DynamicFeeCap) + require.Equal(t, true, a.IsPurgeAttempt) + require.Equal(t, []byte{}, a.Tx.EncodedPayload) + require.Equal(t, *big.NewInt(0), a.Tx.Value) + }) + + t.Run("creates dynamic purge attempt with fields if previous attempt is dynamic", func(t *testing.T) { + n := evmtypes.Nonce(0) + etx := txmgr.Tx{Sequence: &n, FromAddress: addr, EncodedPayload: []byte{1, 2, 3}} + prevAttempt, _, err := cks.NewCustomTxAttempt(ctx, etx, gas.EvmFee{DynamicTipCap: bumpedDynamicTip.Sub(assets.GWei(1)), DynamicFeeCap: bumpedDynamicFee.Sub(assets.GWei(1))}, 100, 0x2, lggr) + require.NoError(t, err) + etx.TxAttempts = append(etx.TxAttempts, prevAttempt) + a, err := cks.NewPurgeTxAttempt(ctx, etx, lggr) + require.NoError(t, err) + // The fee limit is overridden with LimitDefault since purge attempts are just empty attempts + require.Equal(t, gc.limitDefault, a.ChainSpecificFeeLimit) + require.Nil(t, a.TxFee.Legacy) + require.NotNil(t, a.TxFee.DynamicTipCap) + require.NotNil(t, a.TxFee.DynamicFeeCap) + require.Equal(t, bumpedDynamicTip.String(), a.TxFee.DynamicTipCap.String()) + require.Equal(t, bumpedDynamicFee.String(), a.TxFee.DynamicFeeCap.String()) + require.Equal(t, true, a.IsPurgeAttempt) + require.Equal(t, []byte{}, a.Tx.EncodedPayload) + require.Equal(t, *big.NewInt(0), a.Tx.Value) + }) + + t.Run("creates bump purge attempt with fields", func(t *testing.T) { + n := evmtypes.Nonce(0) + etx := txmgr.Tx{Sequence: &n, FromAddress: addr, EncodedPayload: []byte{1, 2, 3}} + prevAttempt, _, err := cks.NewCustomTxAttempt(ctx, etx, gas.EvmFee{Legacy: bumpedLegacy.Sub(assets.GWei(1))}, 100, 0x0, lggr) + require.NoError(t, err) + etx.TxAttempts = append(etx.TxAttempts, prevAttempt) + purgeAttempt, err := cks.NewPurgeTxAttempt(ctx, etx, lggr) + require.NoError(t, err) + etx.TxAttempts = append(etx.TxAttempts, purgeAttempt) + bumpAttempt, _, _, _, err := cks.NewBumpTxAttempt(ctx, etx, purgeAttempt, etx.TxAttempts, lggr) + require.NoError(t, err) + // The fee limit is overridden with LimitDefault since purge attempts are just empty attempts + require.Equal(t, gc.limitDefault, bumpAttempt.ChainSpecificFeeLimit) + require.NotNil(t, bumpAttempt.TxFee.Legacy) + require.Equal(t, bumpedLegacy.String(), bumpAttempt.TxFee.Legacy.String()) + require.Nil(t, bumpAttempt.TxFee.DynamicTipCap) + require.Nil(t, bumpAttempt.TxFee.DynamicFeeCap) + require.Equal(t, true, bumpAttempt.IsPurgeAttempt) + require.Equal(t, []byte{}, bumpAttempt.Tx.EncodedPayload) + require.Equal(t, *big.NewInt(0), bumpAttempt.Tx.Value) + }) +} + func TestTxm_NewCustomTxAttempt_NonRetryableErrors(t *testing.T) { t.Parallel() diff --git a/core/chains/evm/txmgr/builder.go b/core/chains/evm/txmgr/builder.go index 0671f49bb74..caba4d3806c 100644 --- a/core/chains/evm/txmgr/builder.go +++ b/core/chains/evm/txmgr/builder.go @@ -51,7 +51,8 @@ func NewTxm( chainID := txmClient.ConfiguredChainID() evmBroadcaster := NewEvmBroadcaster(txStore, txmClient, txmCfg, feeCfg, txConfig, listenerConfig, keyStore, txAttemptBuilder, lggr, checker, chainConfig.NonceAutoSync()) evmTracker := NewEvmTracker(txStore, keyStore, chainID, lggr) - evmConfirmer := NewEvmConfirmer(txStore, txmClient, txmCfg, feeCfg, txConfig, dbConfig, keyStore, txAttemptBuilder, lggr) + stuckTxDetector := NewStuckTxDetector(lggr, client.ConfiguredChainID(), chainConfig.ChainType(), fCfg.PriceMax(), txConfig.AutoPurge(), estimator, txStore, client) + evmConfirmer := NewEvmConfirmer(txStore, txmClient, txmCfg, feeCfg, txConfig, dbConfig, keyStore, txAttemptBuilder, lggr, stuckTxDetector) var evmResender *Resender if txConfig.ResendAfterThreshold() > 0 { evmResender = NewEvmResender(lggr, txStore, txmClient, evmTracker, keyStore, txmgr.DefaultResenderPollInterval, chainConfig, txConfig) @@ -109,8 +110,9 @@ func NewEvmConfirmer( keystore KeyStore, txAttemptBuilder TxAttemptBuilder, lggr logger.Logger, + stuckTxDetector StuckTxDetector, ) *Confirmer { - return txmgr.NewConfirmer(txStore, client, chainConfig, feeConfig, txConfig, dbConfig, keystore, txAttemptBuilder, lggr, func(r *evmtypes.Receipt) bool { return r == nil }) + return txmgr.NewConfirmer(txStore, client, chainConfig, feeConfig, txConfig, dbConfig, keystore, txAttemptBuilder, lggr, func(r *evmtypes.Receipt) bool { return r == nil }, stuckTxDetector) } // NewEvmTracker instantiates a new EVM tracker for abandoned transactions diff --git a/core/chains/evm/txmgr/confirmer_test.go b/core/chains/evm/txmgr/confirmer_test.go index db2d8a9092f..2ce34505234 100644 --- a/core/chains/evm/txmgr/confirmer_test.go +++ b/core/chains/evm/txmgr/confirmer_test.go @@ -129,7 +129,8 @@ func TestEthConfirmer_Lifecycle(t *testing.T) { ge := config.EVM().GasEstimator() feeEstimator := gas.NewEvmFeeEstimator(lggr, newEst, ge.EIP1559DynamicFees(), ge) txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), ge, ethKeyStore, feeEstimator) - ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), txmgr.NewEvmTxmConfig(config.EVM()), txmgr.NewEvmTxmFeeConfig(ge), config.EVM().Transactions(), gconfig.Database(), ethKeyStore, txBuilder, lggr) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), config.EVM().Transactions().AutoPurge(), feeEstimator, txStore, ethClient) + ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), txmgr.NewEvmTxmConfig(config.EVM()), txmgr.NewEvmTxmFeeConfig(ge), config.EVM().Transactions(), gconfig.Database(), ethKeyStore, txBuilder, lggr, stuckTxDetector) ctx := testutils.Context(t) // Can't close unstarted instance @@ -1645,8 +1646,9 @@ func TestEthConfirmer_RebroadcastWhereNecessary_WithConnectivityCheck(t *testing txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), ge, kst, feeEstimator) addresses := []gethCommon.Address{fromAddress} kst.On("EnabledAddressesForChain", mock.Anything, &cltest.FixtureChainID).Return(addresses, nil).Maybe() + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), ccfg.EVM().Transactions().AutoPurge(), feeEstimator, txStore, ethClient) // Create confirmer with necessary state - ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), ccfg.EVM(), txmgr.NewEvmTxmFeeConfig(ccfg.EVM().GasEstimator()), ccfg.EVM().Transactions(), cfg.Database(), kst, txBuilder, lggr) + ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), ccfg.EVM(), txmgr.NewEvmTxmFeeConfig(ccfg.EVM().GasEstimator()), ccfg.EVM().Transactions(), cfg.Database(), kst, txBuilder, lggr, stuckTxDetector) servicetest.Run(t, ec) currentHead := int64(30) oldEnough := int64(15) @@ -1693,7 +1695,8 @@ func TestEthConfirmer_RebroadcastWhereNecessary_WithConnectivityCheck(t *testing txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), ge, kst, feeEstimator) addresses := []gethCommon.Address{fromAddress} kst.On("EnabledAddressesForChain", mock.Anything, &cltest.FixtureChainID).Return(addresses, nil).Maybe() - ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), ccfg.EVM(), txmgr.NewEvmTxmFeeConfig(ccfg.EVM().GasEstimator()), ccfg.EVM().Transactions(), cfg.Database(), kst, txBuilder, lggr) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), ccfg.EVM().Transactions().AutoPurge(), feeEstimator, txStore, ethClient) + ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), ccfg.EVM(), txmgr.NewEvmTxmFeeConfig(ccfg.EVM().GasEstimator()), ccfg.EVM().Transactions(), cfg.Database(), kst, txBuilder, lggr, stuckTxDetector) servicetest.Run(t, ec) currentHead := int64(30) oldEnough := int64(15) @@ -3119,6 +3122,107 @@ 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.Enabled = ptr(true) + c.EVM[0].Transactions.AutoPurge.Threshold = ptr(autoPurgeThreshold) + c.EVM[0].Transactions.AutoPurge.MinAttempts = 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, "", assets.NewWei(assets.NewEth(100).ToInt()), 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 := mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, nonce, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei)) + + head := evmtypes.Head{ + Hash: utils.NewHash(), + Number: blockNum, + } + ethClient.On("SequenceAt", mock.Anything, mock.Anything, mock.Anything).Return(evmtypes.Nonce(0), nil).Once() + ethClient.On("BatchCallContext", mock.Anything, mock.Anything).Return(nil).Once() + + // First call to ProcessHead should: + // 1. Detect a stuck transaction + // 2. Create a purge attempt for it + // 3. Save the purge attempt to the DB + // 4. Send the purge attempt + err := ec.ProcessHead(ctx, &head) + 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) + + head = evmtypes.Head{ + Hash: utils.NewHash(), + Number: blockNum + 1, + } + 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(blockNum + 1), + TransactionIndex: uint(1), + Status: uint64(1), + } + }).Once() + + // Second call to ProcessHead on next head should: + // 1. Check for receipts for purged transaction + // 2. When receipts are found for a purge attempt, the transaction is marked in the DB as fatal error with error message + err = ec.ProcessHead(ctx, &head) + 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) + require.Equal(t, "transaction terminally stuck", dbTx.Error.String) + }) +} + 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 { @@ -3128,7 +3232,8 @@ func newEthConfirmer(t testing.TB, txStore txmgr.EvmTxStore, ethClient client.Cl return gas.NewFixedPriceEstimator(ge, nil, ge.BlockHistory(), lggr, nil) }, ge.EIP1559DynamicFees(), ge) txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), ge, ks, estimator) - ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), txmgr.NewEvmTxmConfig(config.EVM()), txmgr.NewEvmTxmFeeConfig(ge), config.EVM().Transactions(), gconfig.Database(), ks, txBuilder, lggr) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), config.EVM().Transactions().AutoPurge(), estimator, txStore, ethClient) + ec := txmgr.NewEvmConfirmer(txStore, txmgr.NewEvmTxmClient(ethClient, nil), txmgr.NewEvmTxmConfig(config.EVM()), txmgr.NewEvmTxmFeeConfig(ge), config.EVM().Transactions(), gconfig.Database(), ks, txBuilder, lggr, stuckTxDetector) ec.SetResumeCallback(fn) servicetest.Run(t, ec) return ec diff --git a/core/chains/evm/txmgr/evm_tx_store.go b/core/chains/evm/txmgr/evm_tx_store.go index 505938d3026..ce64f816cd9 100644 --- a/core/chains/evm/txmgr/evm_tx_store.go +++ b/core/chains/evm/txmgr/evm_tx_store.go @@ -48,7 +48,7 @@ type EvmTxStore interface { TxStoreWebApi } -// TxStoreWebApi encapsulates the methods that are not used by the txmgr and only used by the various web controllers and readers +// TxStoreWebApi encapsulates the methods that are not used by the txmgr and only used by the various web controllers, readers, or evm specific components type TxStoreWebApi interface { FindTxAttemptConfirmedByTxIDs(ctx context.Context, ids []int64) ([]TxAttempt, error) FindTxByHash(ctx context.Context, hash common.Hash) (*Tx, error) @@ -57,6 +57,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) + FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state txmgrtypes.TxState, chainID *big.Int) (txs []*Tx, err error) } type TestEvmTxStore interface { @@ -285,6 +286,7 @@ type DbEthTxAttempt struct { TxType int GasTipCap *assets.Wei GasFeeCap *assets.Wei + IsPurgeAttempt bool } func (db *DbEthTxAttempt) FromTxAttempt(attempt *TxAttempt) { @@ -299,6 +301,7 @@ func (db *DbEthTxAttempt) FromTxAttempt(attempt *TxAttempt) { db.TxType = attempt.TxType db.GasTipCap = attempt.TxFee.DynamicTipCap db.GasFeeCap = attempt.TxFee.DynamicFeeCap + db.IsPurgeAttempt = attempt.IsPurgeAttempt // handle state naming difference between generic + EVM if attempt.State == txmgrtypes.TxAttemptInsufficientFunds { @@ -330,6 +333,7 @@ func (db DbEthTxAttempt) ToTxAttempt(attempt *TxAttempt) { DynamicTipCap: db.GasTipCap, DynamicFeeCap: db.GasFeeCap, } + attempt.IsPurgeAttempt = db.IsPurgeAttempt } func dbEthTxAttemptsToEthTxAttempts(dbEthTxAttempt []DbEthTxAttempt) []TxAttempt { @@ -353,8 +357,8 @@ func NewTxStore( } const insertIntoEthTxAttemptsQuery = ` -INSERT INTO evm.tx_attempts (eth_tx_id, gas_price, signed_raw_tx, hash, broadcast_before_block_num, state, created_at, chain_specific_gas_limit, tx_type, gas_tip_cap, gas_fee_cap) -VALUES (:eth_tx_id, :gas_price, :signed_raw_tx, :hash, :broadcast_before_block_num, :state, NOW(), :chain_specific_gas_limit, :tx_type, :gas_tip_cap, :gas_fee_cap) +INSERT INTO evm.tx_attempts (eth_tx_id, gas_price, signed_raw_tx, hash, broadcast_before_block_num, state, created_at, chain_specific_gas_limit, tx_type, gas_tip_cap, gas_fee_cap, is_purge_attempt) +VALUES (:eth_tx_id, :gas_price, :signed_raw_tx, :hash, :broadcast_before_block_num, :state, NOW(), :chain_specific_gas_limit, :tx_type, :gas_tip_cap, :gas_fee_cap, :is_purge_attempt) RETURNING *; ` @@ -822,7 +826,39 @@ ORDER BY evm.txes.nonce ASC, evm.tx_attempts.gas_price DESC, evm.tx_attempts.gas return } -func (o *evmTxStore) SaveFetchedReceipts(ctx context.Context, r []*evmtypes.Receipt, chainID *big.Int) (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.stopCh.Ctx(ctx) + defer cancel() + enabledAddrsBytea := make([][]byte, len(addresses)) + for i, addr := range addresses { + enabledAddrsBytea[i] = addr.Bytes() + } + 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("FindTxsByStateAndFromAddresses failed to load evm.txes: %w", err) + } + if len(dbEtxs) == 0 { + return nil + } + 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 +} + +func (o *evmTxStore) SaveFetchedReceipts(ctx context.Context, r []*evmtypes.Receipt, state txmgrtypes.TxState, errorMsg *string, chainID *big.Int) (err error) { var cancel context.CancelFunc ctx, cancel = o.stopCh.Ctx(ctx) defer cancel() @@ -869,7 +905,7 @@ func (o *evmTxStore) SaveFetchedReceipts(ctx context.Context, r []*evmtypes.Rece valueStrs = append(valueStrs, "(?,?,?,?,?,NOW())") valueArgs = append(valueArgs, r.TxHash, r.BlockHash, r.BlockNumber.Int64(), r.TransactionIndex, receiptJSON) } - valueArgs = append(valueArgs, chainID.String()) + valueArgs = append(valueArgs, state, errorMsg, chainID.String()) /* #nosec G201 */ sql := ` @@ -892,7 +928,7 @@ func (o *evmTxStore) SaveFetchedReceipts(ctx context.Context, r []*evmtypes.Rece RETURNING evm.tx_attempts.eth_tx_id ) UPDATE evm.txes - SET state = 'confirmed' + SET state = ?, error = ? FROM updated_eth_tx_attempts WHERE updated_eth_tx_attempts.eth_tx_id = evm.txes.id AND evm_chain_id = ? diff --git a/core/chains/evm/txmgr/evm_tx_store_test.go b/core/chains/evm/txmgr/evm_tx_store_test.go index 20b39f3a83e..23b8a9fde33 100644 --- a/core/chains/evm/txmgr/evm_tx_store_test.go +++ b/core/chains/evm/txmgr/evm_tx_store_test.go @@ -521,7 +521,7 @@ func TestORM_SaveFetchedReceipts(t *testing.T) { TransactionIndex: uint(1), } - err := txStore.SaveFetchedReceipts(testutils.Context(t), []*evmtypes.Receipt{&txmReceipt}, ethClient.ConfiguredChainID()) + err := txStore.SaveFetchedReceipts(testutils.Context(t), []*evmtypes.Receipt{&txmReceipt}, txmgrcommon.TxConfirmed, nil, ethClient.ConfiguredChainID()) require.NoError(t, err) etx0, err = txStore.FindTxWithAttempts(ctx, etx0.ID) @@ -1194,8 +1194,8 @@ func TestORM_LoadEthTxesAttempts(t *testing.T) { tx, err := db.BeginTx(ctx, nil) require.NoError(t, err) - const insertEthTxAttemptSQL = `INSERT INTO evm.tx_attempts (eth_tx_id, gas_price, signed_raw_tx, hash, broadcast_before_block_num, state, created_at, chain_specific_gas_limit, tx_type, gas_tip_cap, gas_fee_cap) VALUES ( - :eth_tx_id, :gas_price, :signed_raw_tx, :hash, :broadcast_before_block_num, :state, NOW(), :chain_specific_gas_limit, :tx_type, :gas_tip_cap, :gas_fee_cap + const insertEthTxAttemptSQL = `INSERT INTO evm.tx_attempts (eth_tx_id, gas_price, signed_raw_tx, hash, broadcast_before_block_num, state, created_at, chain_specific_gas_limit, tx_type, gas_tip_cap, gas_fee_cap, is_purge_attempt) VALUES ( + :eth_tx_id, :gas_price, :signed_raw_tx, :hash, :broadcast_before_block_num, :state, NOW(), :chain_specific_gas_limit, :tx_type, :gas_tip_cap, :gas_fee_cap, :is_purge_attempt ) RETURNING *` query, args, err := sqlutil.DataSource(db).BindNamed(insertEthTxAttemptSQL, dbAttempt) require.NoError(t, err) diff --git a/core/chains/evm/txmgr/mocks/evm_tx_store.go b/core/chains/evm/txmgr/mocks/evm_tx_store.go index be59d0130a4..465681247e2 100644 --- a/core/chains/evm/txmgr/mocks/evm_tx_store.go +++ b/core/chains/evm/txmgr/mocks/evm_tx_store.go @@ -761,6 +761,36 @@ func (_m *EvmTxStore) FindTxesWithMetaFieldByStates(ctx context.Context, metaFie return r0, r1 } +// FindTxsByStateAndFromAddresses provides a mock function with given fields: ctx, addresses, state, chainID +func (_m *EvmTxStore) FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state types.TxState, chainID *big.Int) ([]*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee], error) { + ret := _m.Called(ctx, addresses, state, chainID) + + if len(ret) == 0 { + panic("no return value specified for FindTxsByStateAndFromAddresses") + } + + var r0 []*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee] + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, []common.Address, types.TxState, *big.Int) ([]*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee], error)); ok { + return rf(ctx, addresses, state, chainID) + } + if rf, ok := ret.Get(0).(func(context.Context, []common.Address, types.TxState, *big.Int) []*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee]); ok { + r0 = rf(ctx, addresses, state, chainID) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee]) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, []common.Address, types.TxState, *big.Int) error); ok { + r1 = rf(ctx, addresses, state, chainID) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // FindTxsRequiringGasBump provides a mock function with given fields: ctx, address, blockNum, gasBumpThreshold, depth, chainID func (_m *EvmTxStore) FindTxsRequiringGasBump(ctx context.Context, address common.Address, blockNum int64, gasBumpThreshold int64, depth int64, chainID *big.Int) ([]*types.Tx[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee], error) { ret := _m.Called(ctx, address, blockNum, gasBumpThreshold, depth, chainID) @@ -1135,17 +1165,17 @@ func (_m *EvmTxStore) SaveConfirmedMissingReceiptAttempt(ctx context.Context, ti return r0 } -// SaveFetchedReceipts provides a mock function with given fields: ctx, receipts, chainID -func (_m *EvmTxStore) SaveFetchedReceipts(ctx context.Context, receipts []*evmtypes.Receipt, chainID *big.Int) error { - ret := _m.Called(ctx, receipts, chainID) +// SaveFetchedReceipts provides a mock function with given fields: ctx, r, state, errorMsg, chainID +func (_m *EvmTxStore) SaveFetchedReceipts(ctx context.Context, r []*evmtypes.Receipt, state types.TxState, errorMsg *string, chainID *big.Int) error { + ret := _m.Called(ctx, r, state, errorMsg, chainID) if len(ret) == 0 { panic("no return value specified for SaveFetchedReceipts") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, []*evmtypes.Receipt, *big.Int) error); ok { - r0 = rf(ctx, receipts, chainID) + if rf, ok := ret.Get(0).(func(context.Context, []*evmtypes.Receipt, types.TxState, *string, *big.Int) error); ok { + r0 = rf(ctx, r, state, errorMsg, chainID) } else { r0 = ret.Error(0) } diff --git a/core/chains/evm/txmgr/models.go b/core/chains/evm/txmgr/models.go index be06f5dd5e9..f8682ffd500 100644 --- a/core/chains/evm/txmgr/models.go +++ b/core/chains/evm/txmgr/models.go @@ -38,6 +38,7 @@ type ( TxAttempt = txmgrtypes.TxAttempt[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee] Receipt = dbReceipt // EvmReceipt is the exported DB table model for receipts ReceiptPlus = txmgrtypes.ReceiptPlus[*evmtypes.Receipt] + StuckTxDetector = txmgrtypes.StuckTxDetector[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee] TxmClient = txmgrtypes.TxmClient[*big.Int, common.Address, common.Hash, common.Hash, *evmtypes.Receipt, evmtypes.Nonce, gas.EvmFee] TransactionClient = txmgrtypes.TransactionClient[*big.Int, common.Address, common.Hash, common.Hash, evmtypes.Nonce, gas.EvmFee] ChainReceipt = txmgrtypes.ChainReceipt[common.Hash, common.Hash] diff --git a/core/chains/evm/txmgr/stuck_tx_detector.go b/core/chains/evm/txmgr/stuck_tx_detector.go new file mode 100644 index 00000000000..d48cdf00e79 --- /dev/null +++ b/core/chains/evm/txmgr/stuck_tx_detector.go @@ -0,0 +1,375 @@ +package txmgr + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "math/big" + "net/http" + "net/url" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/rpc" + + "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" +) + +type stuckTxDetectorGasEstimator interface { + GetFee(ctx context.Context, calldata []byte, feeLimit uint64, maxFeePrice *assets.Wei, opts ...feetypes.Opt) (fee gas.EvmFee, chainSpecificFeeLimit uint64, err error) +} + +type stuckTxDetectorClient interface { + BatchCallContext(ctx context.Context, b []rpc.BatchElem) error +} + +type stuckTxDetectorTxStore interface { + FindTxsByStateAndFromAddresses(ctx context.Context, addresses []common.Address, state types.TxState, chainID *big.Int) (txs []*Tx, err error) +} + +type stuckTxDetectorConfig interface { + Enabled() bool + Threshold() uint32 + MinAttempts() uint32 + DetectionApiUrl() *url.URL +} + +type stuckTxDetector struct { + lggr logger.Logger + chainID *big.Int + chainType config.ChainType + maxPrice *assets.Wei + cfg stuckTxDetectorConfig + + gasEstimator stuckTxDetectorGasEstimator + txStore stuckTxDetectorTxStore + chainClient stuckTxDetectorClient + httpClient *http.Client + + purgeBlockNumLock sync.RWMutex + purgeBlockNumMap map[common.Address]int64 // Tracks the last block num a tx was purged for each from address if the PurgeOverflowTxs feature is enabled +} + +func NewStuckTxDetector(lggr logger.Logger, chainID *big.Int, chainType config.ChainType, maxPrice *assets.Wei, cfg stuckTxDetectorConfig, gasEstimator stuckTxDetectorGasEstimator, txStore stuckTxDetectorTxStore, chainClient stuckTxDetectorClient) *stuckTxDetector { + t := http.DefaultTransport.(*http.Transport).Clone() + t.DisableCompression = true + httpClient := &http.Client{Transport: t} + return &stuckTxDetector{ + lggr: lggr, + chainID: chainID, + chainType: chainType, + maxPrice: maxPrice, + cfg: cfg, + gasEstimator: gasEstimator, + txStore: txStore, + chainClient: chainClient, + httpClient: httpClient, + purgeBlockNumMap: make(map[common.Address]int64), + } +} + +func (d *stuckTxDetector) LoadPurgeBlockNumMap(ctx context.Context, addresses []common.Address) error { + // Skip loading purge block num map if auto-purge feature disabled or Threshold is set to 0 + if !d.cfg.Enabled() || d.cfg.Threshold() == 0 { + return nil + } + 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 auto-purge 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) { + if !d.cfg.Enabled() { + return nil, nil + } + txs, err := d.FindUnconfirmedTxWithLowestNonce(ctx, enabledAddresses) + if err != nil { + return nil, fmt.Errorf("failed to get list of transactions waiting confirmations with lowest nonce for distinct from addresses: %w", err) + } + // No transactions found + if len(txs) == 0 { + return nil, nil + } + + switch d.chainType { + case config.ChainScroll: + return d.detectStuckTransactionsScroll(ctx, txs) + case config.ChainZkEvm: + return d.detectStuckTransactionsZkEVM(ctx, txs) + default: + return d.detectStuckTransactionsHeuristic(ctx, txs, blockNum) + } +} + +// Finds the lowest nonce Unconfirmed transaction for each enabled address +// Only the earliest transaction can be considered terminally stuck. All others may be valid and just stuck behind the nonce +func (d *stuckTxDetector) FindUnconfirmedTxWithLowestNonce(ctx context.Context, enabledAddresses []common.Address) ([]Tx, error) { + // Loads attempts within tx + 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 + } else if lowestNonceTx := lowestNonceTxMap[tx.FromAddress]; *lowestNonceTx.Sequence > *tx.Sequence { + lowestNonceTxMap[tx.FromAddress] = *tx + } + } + + // Build list of potentially stuck tx but exclude any that are already marked for purge + var stuckTxs []Tx + for _, tx := range lowestNonceTxMap { + // Attempts are loaded newest to oldest so one marked for purge will always be first + if len(tx.TxAttempts) > 0 && !tx.TxAttempts[0].IsPurgeAttempt { + stuckTxs = append(stuckTxs, tx) + } + } + + return stuckTxs, nil +} + +// Uses a heuristic to determine a stuck transaction potentially due to overflow +// This method can be unreliable and may result in false positives but it is best effort to keep the TXM from getting blocked +// 1. Check if Threshold amount of blocks have passed since the last purge of a tx for the same fromAddress +// 2. If 1 is true, check if Threshold amount of blocks have passed since the initial broadcast +// 3. If 2 is true, check if the transaction has at least MinAttempts amount of broadcasted attempts +// 4. If 3 is true, check if the latest attempt's gas price is higher than what our gas estimator's GetFee method returns +// 5. If 4 is true, the transaction is likely stuck due to overflow +func (d *stuckTxDetector) detectStuckTransactionsHeuristic(ctx context.Context, txs []Tx, blockNum int64) ([]Tx, error) { + d.purgeBlockNumLock.RLock() + defer d.purgeBlockNumLock.RUnlock() + // Get gas price from internal gas estimator + // Send with max gas price time 2 to prevent the results from being capped. Need the market gas price here. + marketGasPrice, _, err := d.gasEstimator.GetFee(ctx, []byte{}, 0, d.maxPrice.Mul(big.NewInt(2))) + if err != nil { + return txs, fmt.Errorf("failed to get market gas price for overflow detection: %w", err) + } + var stuckTxs []Tx + for _, tx := range txs { + // 1. Check if Threshold amount of blocks have passed since the last purge of a tx for the same fromAddress + // Used to rate limit purging to prevent a potential valid tx that was stuck behind an overflow tx from also getting purged without having enough time to be confirmed + d.purgeBlockNumLock.RLock() + lastPurgeBlockNum := d.purgeBlockNumMap[tx.FromAddress] + d.purgeBlockNumLock.RUnlock() + if lastPurgeBlockNum > blockNum-int64(d.cfg.Threshold()) { + continue + } + // Tx attempts are loaded from newest to oldest + oldestBroadcastAttempt, newestBroadcastAttempt, broadcastedAttemptsCount := findBroadcastedAttempts(tx) + // 2. Check if Threshold amount of blocks have passed since the oldest attempt's broadcast block num + if *oldestBroadcastAttempt.BroadcastBeforeBlockNum > blockNum-int64(d.cfg.Threshold()) { + continue + } + // 3. Check if the transaction has at least MinAttempts amount of broadcasted attempts + if broadcastedAttemptsCount < d.cfg.MinAttempts() { + continue + } + // 4. Check if the newest broadcasted attempt's gas price is higher than what our gas estimator's GetFee method returns + if compareGasFees(newestBroadcastAttempt.TxFee, marketGasPrice) <= 0 { + continue + } + // 5. Return the transaction since it is likely stuck due to overflow + stuckTxs = append(stuckTxs, tx) + } + return stuckTxs, nil +} + +func compareGasFees(attemptGas gas.EvmFee, marketGas gas.EvmFee) int { + if attemptGas.Legacy != nil && marketGas.Legacy != nil { + return attemptGas.Legacy.Cmp(marketGas.Legacy) + } + if attemptGas.DynamicFeeCap.Cmp(marketGas.DynamicFeeCap) == 0 { + return attemptGas.DynamicTipCap.Cmp(marketGas.DynamicTipCap) + } + return attemptGas.DynamicFeeCap.Cmp(marketGas.DynamicFeeCap) +} + +// Assumes tx attempts are loaded newest to oldest +func findBroadcastedAttempts(tx Tx) (oldestAttempt TxAttempt, newestAttempt TxAttempt, broadcastedCount uint32) { + foundNewest := false + for _, attempt := range tx.TxAttempts { + if attempt.State != types.TxAttemptBroadcast { + continue + } + if !foundNewest { + newestAttempt = attempt + foundNewest = true + } + oldestAttempt = attempt + broadcastedCount++ + } + return +} + +type scrollRequest struct { + Txs []string `json:"txs"` +} + +type scrollResponse struct { + Errcode int `json:"errcode"` + Errmsg string `json:"errmsg"` + Data map[string]int `json:"data"` +} + +// Uses the custom Scroll skipped endpoint to determine an overflow transaction +func (d *stuckTxDetector) detectStuckTransactionsScroll(ctx context.Context, txs []Tx) ([]Tx, error) { + if d.cfg.DetectionApiUrl() == nil { + return nil, fmt.Errorf("expected DetectionApiUrl config to be set for chain type: %s", d.chainType) + } + + attemptHashMap := make(map[string]Tx) + + request := new(scrollRequest) + // Populate the request with the tx hash of the latest broadcast attempt from every tx + for _, tx := range txs { + for _, attempt := range tx.TxAttempts { + if attempt.State == types.TxAttemptBroadcast { + request.Txs = append(request.Txs, attempt.Hash.String()) + attemptHashMap[attempt.Hash.String()] = tx + break + } + } + } + jsonReq, err := json.Marshal(request) + if err != nil { + return nil, fmt.Errorf("failed to marshal json request %v for custom endpoint: %w", request, err) + } + + // Build http post request + url := fmt.Sprintf("%s/v1/sequencer/tx/skipped", d.cfg.DetectionApiUrl()) + bodyReader := bytes.NewReader(jsonReq) + postReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bodyReader) + if err != nil { + return nil, fmt.Errorf("failed to make new request with context: %w", err) + } + // Send request + resp, err := d.httpClient.Do(postReq) + if err != nil { + return nil, fmt.Errorf("request to scroll's custom endpoint failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return nil, fmt.Errorf("request failed with status %d", resp.StatusCode) + } + // Decode the response into expected type + scrollResp := new(scrollResponse) + err = json.NewDecoder(resp.Body).Decode(scrollResp) + if err != nil { + return nil, fmt.Errorf("failed to unmarshal response into struct: %w", err) + } + if scrollResp.Errcode != 0 || scrollResp.Errmsg != "" { + return nil, fmt.Errorf("scroll's custom endpoint returned an error with code: %d, message: %s", scrollResp.Errcode, scrollResp.Errmsg) + } + + // Return all transactions marked with status 1 signaling they have been skipped due to overflow + var stuckTx []Tx + for hash, status := range scrollResp.Data { + if status == 1 { + stuckTx = append(stuckTx, attemptHashMap[hash]) + } + } + + return stuckTx, nil +} + +// Uses eth_getTransactionByHash to detect that a transaction has been discarded due to overflow +// Currently only used by zkEVM but if other chains follow the same behavior in the future +func (d *stuckTxDetector) detectStuckTransactionsZkEVM(ctx context.Context, txs []Tx) ([]Tx, error) { + txReqs := make([]rpc.BatchElem, len(txs)) + txHashMap := make(map[common.Hash]Tx) + txRes := make([]*map[string]interface{}, len(txs)) + + // Build batch request elems to perform + // Does not need to be separated out into smaller batches + // Max number of transactions to check is equal to the number of enabled addresses which is a relatively small amount + for i, tx := range txs { + latestAttemptHash := tx.TxAttempts[0].Hash + var result map[string]interface{} + txReqs[i] = rpc.BatchElem{ + Method: "eth_getTransactionByHash", + Args: []interface{}{ + latestAttemptHash, + }, + Result: &result, + } + txHashMap[latestAttemptHash] = tx + txRes[i] = &result + } + + // Send batch request + err := d.chainClient.BatchCallContext(ctx, txReqs) + if err != nil { + return nil, fmt.Errorf("failed to get transactions by hash in batch: %w", err) + } + + // Parse results to find tx skipped due to zk overflow + // If the result is nil, the transaction was discarded due to overflow + var stuckTxs []Tx + for i, req := range txReqs { + txHash := req.Args[0].(common.Hash) + if req.Error != nil { + d.lggr.Debugf("failed to get transaction by hash (%s): %w", txHash.String(), req.Error) + continue + } + result := *txRes[i] + if result == nil { + tx := txHashMap[txHash] + stuckTxs = append(stuckTxs, tx) + } + } + return stuckTxs, nil +} + +// Once a purged tx's empty attempt is confirmed, this method is used to set at which block num the tx was purged at for the fromAddress +func (d *stuckTxDetector) SetPurgeBlockNum(fromAddress common.Address, blockNum int64) { + d.purgeBlockNumLock.Lock() + defer d.purgeBlockNumLock.Unlock() + d.purgeBlockNumMap[fromAddress] = blockNum +} + +func (d *stuckTxDetector) StuckTxFatalError() *string { + var errorMsg string + switch d.chainType { + case config.ChainScroll, config.ChainZkEvm: + errorMsg = "transaction skipped by chain" + default: + errorMsg = "transaction terminally stuck" + } + + return &errorMsg +} diff --git a/core/chains/evm/txmgr/stuck_tx_detector_test.go b/core/chains/evm/txmgr/stuck_tx_detector_test.go new file mode 100644 index 00000000000..39c275d286f --- /dev/null +++ b/core/chains/evm/txmgr/stuck_tx_detector_test.go @@ -0,0 +1,433 @@ +package txmgr_test + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/rpc" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "gopkg.in/guregu/null.v4" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + commonconfig "github.com/smartcontractkit/chainlink/v2/common/config" + txmgrcommon "github.com/smartcontractkit/chainlink/v2/common/txmgr" + txmgrtypes "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" + gasmocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/gas/mocks" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types" + "github.com/smartcontractkit/chainlink/v2/core/internal/cltest" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/evmtest" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest" +) + +var ( + tenGwei = assets.NewWeiI(10_000_000_000) + oneGwei = assets.NewWeiI(1_000_000_000) +) + +func TestStuckTxDetector_Disabled(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) + + lggr := logger.Test(t) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + autoPurgeCfg := testAutoPurgeConfig{ + enabled: false, + } + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), autoPurgeCfg, feeEstimator, txStore, ethClient) + + t.Run("returns empty list if auto-purge feature is disabled", func(t *testing.T) { + txs, err := stuckTxDetector.DetectStuckTransactions(testutils.Context(t), []common.Address{fromAddress}, 100) + require.NoError(t, err) + require.Len(t, txs, 0) + }) +} + +func TestStuckTxDetector_LoadPurgeBlockNumMap(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + txStore := cltest.NewTestTxStore(t, db) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + ctx := testutils.Context(t) + blockNum := int64(100) + + lggr := logger.Test(t) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + marketGasPrice := assets.GWei(15) + fee := gas.EvmFee{Legacy: marketGasPrice} + feeEstimator.On("GetFee", mock.Anything, []byte{}, uint64(0), mock.Anything).Return(fee, uint64(0), nil) + autoPurgeThreshold := uint32(5) + autoPurgeMinAttempts := uint32(3) + autoPurgeCfg := testAutoPurgeConfig{ + enabled: true, // Enable auto-purge feature for testing + threshold: autoPurgeThreshold, + minAttempts: autoPurgeMinAttempts, + } + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), autoPurgeCfg, feeEstimator, txStore, ethClient) + + t.Run("purge num map loaded on startup rate limits new purges on startup", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + mustInsertFatalErrorTxWithError(t, txStore, 0, fromAddress, blockNum) + + err := stuckTxDetector.LoadPurgeBlockNumMap(ctx, []common.Address{fromAddress}) + require.NoError(t, err) + + enabledAddresses := []common.Address{fromAddress} + // Create attempts broadcasted autoPurgeThreshold block ago to ensure broadcast block num check is not being triggered + // Create autoPurgeMinAttempts number of attempts to ensure the broadcast attempt count check is not being triggered + // Create attempts so that the latest has a higher gas price than the market to ensure the gas price check is not being triggered + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 1, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei)) + + // Run detection logic on autoPurgeThreshold blocks past the latest broadcast attempt + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) +} + +func TestStuckTxDetector_FindPotentialStuckTxs(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + _, config := newTestChainScopedConfig(t) + txStore := cltest.NewTestTxStore(t, db) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + ctx := testutils.Context(t) + + lggr := logger.Test(t) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), config.EVM().Transactions().AutoPurge(), feeEstimator, txStore, ethClient) + + t.Run("returns empty list if no unconfimed transactions found", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + stuckTxs, err := stuckTxDetector.FindUnconfirmedTxWithLowestNonce(ctx, []common.Address{fromAddress}) + require.NoError(t, err) + require.Len(t, stuckTxs, 0) + }) + + t.Run("returns 1 unconfirmed transaction for each unique from address", func(t *testing.T) { + _, fromAddress1 := cltest.MustInsertRandomKey(t, ethKeyStore) + _, fromAddress2 := cltest.MustInsertRandomKey(t, ethKeyStore) + // Insert 2 txs for from address, should only return the lowest nonce txs + cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 0, fromAddress1) + cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 1, fromAddress1) + // Insert 1 tx for other from address, should return a tx + cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 0, fromAddress2) + stuckTxs, err := stuckTxDetector.FindUnconfirmedTxWithLowestNonce(ctx, []common.Address{fromAddress1, fromAddress2}) + require.NoError(t, err) + + require.Len(t, stuckTxs, 2) + var foundFromAddresses []common.Address + for _, stuckTx := range stuckTxs { + // Make sure lowest nonce tx is returned for both from addresses + require.Equal(t, types.Nonce(0), *stuckTx.Sequence) + // Make sure attempts are loaded into the tx + require.Len(t, stuckTx.TxAttempts, 1) + foundFromAddresses = append(foundFromAddresses, stuckTx.FromAddress) + } + require.Contains(t, foundFromAddresses, fromAddress1) + require.Contains(t, foundFromAddresses, fromAddress2) + }) + + t.Run("excludes transactions already marked for purge", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + mustInsertUnconfirmedEthTxWithBroadcastPurgeAttempt(t, txStore, 0, fromAddress) + stuckTxs, err := stuckTxDetector.FindUnconfirmedTxWithLowestNonce(ctx, []common.Address{fromAddress}) + require.NoError(t, err) + require.Len(t, stuckTxs, 0) + }) +} + +func TestStuckTxDetector_DetectStuckTransactionsHeuristic(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + txStore := cltest.NewTestTxStore(t, db) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + ctx := testutils.Context(t) + + lggr := logger.Test(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + // Return 10 gwei as market gas price + marketGasPrice := tenGwei + fee := gas.EvmFee{Legacy: marketGasPrice} + feeEstimator.On("GetFee", mock.Anything, []byte{}, uint64(0), mock.Anything).Return(fee, uint64(0), nil) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + autoPurgeThreshold := uint32(5) + autoPurgeMinAttempts := uint32(3) + autoPurgeCfg := testAutoPurgeConfig{ + enabled: true, // Enable auto-purge feature for testing + threshold: autoPurgeThreshold, + minAttempts: autoPurgeMinAttempts, + } + blockNum := int64(100) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, "", assets.NewWei(assets.NewEth(100).ToInt()), autoPurgeCfg, feeEstimator, txStore, ethClient) + + t.Run("not stuck, Threshold amount of blocks have not passed since broadcast", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + enabledAddresses := []common.Address{fromAddress} + // Create attempts broadcasted at the current broadcast number to test the block num threshold check + // Create autoPurgeMinAttempts number of attempts to ensure the broadcast attempt count check is not being triggered + // Create attempts so that the latest has a higher gas price than the market to ensure the gas price check is not being triggered + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, autoPurgeMinAttempts, blockNum, marketGasPrice.Add(oneGwei)) + + // Run detection logic on the same block number as the latest broadcast attempt to stay within the autoPurgeThreshold + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) + + t.Run("not stuck, Threshold amount of blocks have not passed since last purge", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + enabledAddresses := []common.Address{fromAddress} + // Create attempts broadcasted autoPurgeThreshold block ago to ensure broadcast block num check is not being triggered + // Create autoPurgeMinAttempts number of attempts to ensure the broadcast attempt count check is not being triggered + // Create attempts so that the latest has a higher gas price than the market to ensure the gas price check is not being triggered + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei)) + + // Set the last purge block num as the current block num to test rate limiting condition + stuckTxDetector.SetPurgeBlockNum(fromAddress, blockNum) + + // Run detection logic on autoPurgeThreshold blocks past the latest broadcast attempt + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) + + t.Run("not stuck, MinAttempts amount of attempts have not been broadcasted", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + enabledAddresses := []common.Address{fromAddress} + // Create attempts broadcasted autoPurgeThreshold block ago to ensure broadcast block num check is not being triggered + // Create fewer attempts than autoPurgeMinAttempts to test min attempt check + // Create attempts so that the latest has a higher gas price than the market to ensure the gas price check is not being triggered + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, autoPurgeMinAttempts-1, blockNum-int64(autoPurgeThreshold), marketGasPrice.Add(oneGwei)) + + // Run detection logic on autoPurgeThreshold blocks past the latest broadcast attempt + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) + + t.Run("not stuck, transaction gas price is lower than market gas price", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + enabledAddresses := []common.Address{fromAddress} + // Create attempts broadcasted autoPurgeThreshold block ago to ensure broadcast block num check is not being triggered + // Create autoPurgeMinAttempts number of attempts to ensure the broadcast attempt count check is not being triggered + // Create attempts so that the latest has a lower gas price than the market to test the gas price check + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold), marketGasPrice.Sub(oneGwei)) + + // Run detection logic on autoPurgeThreshold blocks past the latest broadcast attempt + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) + + t.Run("detects stuck transaction", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + enabledAddresses := []common.Address{fromAddress} + // 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 + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, autoPurgeMinAttempts, blockNum-int64(autoPurgeThreshold)+int64(autoPurgeMinAttempts-1), marketGasPrice.Add(oneGwei)) + + // Run detection logic on autoPurgeThreshold blocks past the latest broadcast attempt + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, enabledAddresses, blockNum) + require.NoError(t, err) + require.Len(t, txs, 1) + }) +} + +func TestStuckTxDetector_DetectStuckTransactionsZkEVM(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + txStore := cltest.NewTestTxStore(t, db) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + ctx := testutils.Context(t) + + lggr := logger.Test(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + autoPurgeCfg := testAutoPurgeConfig{ + enabled: true, + } + blockNum := int64(100) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, commonconfig.ChainZkEvm, assets.NewWei(assets.NewEth(100).ToInt()), autoPurgeCfg, feeEstimator, txStore, ethClient) + t.Run("returns empty list if no stuck transactions identified", func(t *testing.T) { + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + tx := mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress, 1, blockNum, tenGwei) + attempts := tx.TxAttempts[0] + // Request still returns transaction by hash, transaction not discarded by network and not considered stuck + ethClient.On("BatchCallContext", mock.Anything, mock.MatchedBy(func(b []rpc.BatchElem) bool { + return len(b) == 1 && cltest.BatchElemMatchesParams(b[0], attempts.Hash, "eth_getTransactionByHash") + })).Return(nil).Run(func(args mock.Arguments) { + elems := args.Get(1).([]rpc.BatchElem) + resp, err := json.Marshal(types.Transaction{}) + require.NoError(t, err) + elems[0].Error = json.Unmarshal(resp, elems[0].Result) + }).Once() + + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, []common.Address{fromAddress}, blockNum) + require.NoError(t, err) + require.Len(t, txs, 0) + }) + + t.Run("returns stuck transactions discarded by chain", func(t *testing.T) { + // Insert tx that will be mocked as stuck + _, fromAddress1 := cltest.MustInsertRandomKey(t, ethKeyStore) + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress1, 1, blockNum, tenGwei) + + // Insert tx that will still be valid + _, fromAddress2 := cltest.MustInsertRandomKey(t, ethKeyStore) + mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress2, 1, blockNum, tenGwei) + + // Return nil response for a tx and a normal response for the other + ethClient.On("BatchCallContext", mock.Anything, mock.MatchedBy(func(b []rpc.BatchElem) bool { + return len(b) == 2 + })).Return(nil).Run(func(args mock.Arguments) { + elems := args.Get(1).([]rpc.BatchElem) + elems[0].Result = nil // Return nil to signal discarded tx + resp, err := json.Marshal(types.Transaction{}) + require.NoError(t, err) + elems[1].Error = json.Unmarshal(resp, elems[1].Result) // Return non-nil result to signal a valid tx + }).Once() + + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, []common.Address{fromAddress1, fromAddress2}, blockNum) + require.NoError(t, err) + // Expect only 1 tx to return as stuck due to nil eth_getTransactionByHash response + require.Len(t, txs, 1) + }) +} + +func TestStuckTxDetector_DetectStuckTransactionsScroll(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + txStore := cltest.NewTestTxStore(t, db) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + ctx := testutils.Context(t) + + lggr := logger.Test(t) + feeEstimator := gasmocks.NewEvmFeeEstimator(t) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + blockNum := int64(100) + + t.Run("returns stuck tx identified using the custom scroll API", func(t *testing.T) { + // Insert tx that will be mocked as stuck + _, fromAddress1 := cltest.MustInsertRandomKey(t, ethKeyStore) + tx1 := mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress1, 1, blockNum, tenGwei) + attempts1 := tx1.TxAttempts[0] + + // Insert tx that will still be valid + _, fromAddress2 := cltest.MustInsertRandomKey(t, ethKeyStore) + tx2 := mustInsertUnconfirmedTxWithBroadcastAttempts(t, txStore, 0, fromAddress2, 1, blockNum, tenGwei) + attempts2 := tx2.TxAttempts[0] + + testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) { + _, err := res.Write([]byte(fmt.Sprintf(`{"errcode": 0,"errmsg": "","data": {"%s": 1, "%s": 0}}`, attempts1.Hash, attempts2.Hash))) + require.NoError(t, err) + })) + defer func() { testServer.Close() }() + testUrl, err := url.Parse(testServer.URL) + require.NoError(t, err) + + autoPurgeCfg := testAutoPurgeConfig{ + enabled: true, + detectionApiUrl: testUrl, + } + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, testutils.FixtureChainID, commonconfig.ChainScroll, assets.NewWei(assets.NewEth(100).ToInt()), autoPurgeCfg, feeEstimator, txStore, ethClient) + + txs, err := stuckTxDetector.DetectStuckTransactions(ctx, []common.Address{fromAddress1, fromAddress2}, blockNum) + require.NoError(t, err) + require.Len(t, txs, 1) + require.Equal(t, tx1.ID, txs[0].ID) + }) +} + +func mustInsertUnconfirmedTxWithBroadcastAttempts(t *testing.T, txStore txmgr.TestEvmTxStore, nonce int64, fromAddress common.Address, numAttempts uint32, latestBroadcastBlockNum int64, latestGasPrice *assets.Wei) txmgr.Tx { + ctx := testutils.Context(t) + etx := cltest.MustInsertUnconfirmedEthTx(t, txStore, nonce, fromAddress) + // Insert attempts from oldest to newest + for i := int64(numAttempts - 1); i >= 0; i-- { + blockNum := latestBroadcastBlockNum - i + attempt := cltest.NewLegacyEthTxAttempt(t, etx.ID) + + attempt.State = txmgrtypes.TxAttemptBroadcast + attempt.BroadcastBeforeBlockNum = &blockNum + attempt.TxFee = gas.EvmFee{Legacy: latestGasPrice.Sub(assets.NewWeiI(i))} + require.NoError(t, txStore.InsertTxAttempt(ctx, &attempt)) + } + etx, err := txStore.FindTxWithAttempts(ctx, etx.ID) + require.NoError(t, err) + return etx +} + +func mustInsertFatalErrorTxWithError(t *testing.T, txStore txmgr.TestEvmTxStore, nonce int64, fromAddress common.Address, blockNum int64) txmgr.Tx { + etx := cltest.NewEthTx(fromAddress) + etx.State = txmgrcommon.TxFatalError + etx.Error = null.StringFrom("fatal error") + broadcastAt := time.Now() + etx.BroadcastAt = &broadcastAt + etx.InitialBroadcastAt = &broadcastAt + n := types.Nonce(nonce) + etx.Sequence = &n + etx.ChainID = testutils.FixtureChainID + require.NoError(t, txStore.InsertTx(testutils.Context(t), &etx)) + + attempt := cltest.NewLegacyEthTxAttempt(t, etx.ID) + ctx := testutils.Context(t) + attempt.State = txmgrtypes.TxAttemptBroadcast + attempt.IsPurgeAttempt = true + require.NoError(t, txStore.InsertTxAttempt(ctx, &attempt)) + + receipt := newTxReceipt(attempt.Hash, int(blockNum), 0) + _, err := txStore.InsertReceipt(ctx, &receipt) + require.NoError(t, err) + + etx, err = txStore.FindTxWithAttempts(ctx, etx.ID) + require.NoError(t, err) + return etx +} + +func mustInsertUnconfirmedEthTxWithBroadcastPurgeAttempt(t *testing.T, txStore txmgr.TestEvmTxStore, nonce int64, fromAddress common.Address) txmgr.Tx { + etx := cltest.MustInsertUnconfirmedEthTx(t, txStore, nonce, fromAddress) + attempt := cltest.NewLegacyEthTxAttempt(t, etx.ID) + ctx := testutils.Context(t) + + attempt.State = txmgrtypes.TxAttemptBroadcast + attempt.IsPurgeAttempt = true + require.NoError(t, txStore.InsertTxAttempt(ctx, &attempt)) + etx, err := txStore.FindTxWithAttempts(ctx, etx.ID) + require.NoError(t, err) + return etx +} + +type testAutoPurgeConfig struct { + enabled bool + threshold uint32 + minAttempts uint32 + detectionApiUrl *url.URL +} + +func (t testAutoPurgeConfig) Enabled() bool { return t.enabled } +func (t testAutoPurgeConfig) Threshold() uint32 { return t.threshold } +func (t testAutoPurgeConfig) MinAttempts() uint32 { return t.minAttempts } +func (t testAutoPurgeConfig) DetectionApiUrl() *url.URL { return t.detectionApiUrl } diff --git a/core/chains/evm/txmgr/test_helpers.go b/core/chains/evm/txmgr/test_helpers.go index 64d23373282..ea19e056431 100644 --- a/core/chains/evm/txmgr/test_helpers.go +++ b/core/chains/evm/txmgr/test_helpers.go @@ -1,6 +1,7 @@ package txmgr import ( + "net/url" "testing" "time" @@ -48,16 +49,22 @@ type TestEvmConfig struct { ResendAfterThreshold time.Duration BumpThreshold uint64 MaxQueued uint64 + Enabled bool + Threshold uint32 + MinAttempts uint32 + DetectionApiUrl *url.URL } func (e *TestEvmConfig) Transactions() evmconfig.Transactions { - return &transactionsConfig{e: e} + return &transactionsConfig{e: e, autoPurge: &autoPurgeConfig{}} } func (e *TestEvmConfig) NonceAutoSync() bool { return true } func (e *TestEvmConfig) FinalityDepth() uint32 { return 42 } +func (e *TestEvmConfig) ChainType() commonconfig.ChainType { return "" } + type TestGasEstimatorConfig struct { bumpThreshold uint64 } @@ -115,15 +122,23 @@ func (b *TestBlockHistoryConfig) TransactionPercentile() uint16 { return 42 type transactionsConfig struct { evmconfig.Transactions - e *TestEvmConfig + e *TestEvmConfig + autoPurge evmconfig.AutoPurgeConfig +} + +func (*transactionsConfig) ForwardersEnabled() bool { return true } +func (t *transactionsConfig) MaxInFlight() uint32 { return t.e.MaxInFlight } +func (t *transactionsConfig) MaxQueued() uint64 { return t.e.MaxQueued } +func (t *transactionsConfig) ReaperInterval() time.Duration { return t.e.ReaperInterval } +func (t *transactionsConfig) ReaperThreshold() time.Duration { return t.e.ReaperThreshold } +func (t *transactionsConfig) ResendAfterThreshold() time.Duration { return t.e.ResendAfterThreshold } +func (t *transactionsConfig) AutoPurge() evmconfig.AutoPurgeConfig { return t.autoPurge } + +type autoPurgeConfig struct { + evmconfig.AutoPurgeConfig } -func (*transactionsConfig) ForwardersEnabled() bool { return true } -func (t *transactionsConfig) MaxInFlight() uint32 { return t.e.MaxInFlight } -func (t *transactionsConfig) MaxQueued() uint64 { return t.e.MaxQueued } -func (t *transactionsConfig) ReaperInterval() time.Duration { return t.e.ReaperInterval } -func (t *transactionsConfig) ReaperThreshold() time.Duration { return t.e.ReaperThreshold } -func (t *transactionsConfig) ResendAfterThreshold() time.Duration { return t.e.ResendAfterThreshold } +func (a *autoPurgeConfig) Enabled() bool { return false } type MockConfig struct { EvmConfig *TestEvmConfig diff --git a/core/chains/evm/txmgr/txmgr_test.go b/core/chains/evm/txmgr/txmgr_test.go index 85d25d8a70b..b0823c99705 100644 --- a/core/chains/evm/txmgr/txmgr_test.go +++ b/core/chains/evm/txmgr/txmgr_test.go @@ -641,7 +641,7 @@ func mustInsertConfirmedEthTxBySaveFetchedReceipts(t *testing.T, txStore txmgr.T BlockNumber: big.NewInt(nonce), TransactionIndex: uint(1), } - err := txStore.SaveFetchedReceipts(testutils.Context(t), []*evmtypes.Receipt{&receipt}, &chainID) + err := txStore.SaveFetchedReceipts(testutils.Context(t), []*evmtypes.Receipt{&receipt}, txmgrcommon.TxConfirmed, nil, &chainID) require.NoError(t, err) return etx } diff --git a/core/cmd/shell_local.go b/core/cmd/shell_local.go index 7c9c025d4be..780105d9868 100644 --- a/core/cmd/shell_local.go +++ b/core/cmd/shell_local.go @@ -659,8 +659,9 @@ func (s *Shell) RebroadcastTransactions(c *cli.Context) (err error) { txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), chain.Config().EVM().GasEstimator(), keyStore.Eth(), nil) cfg := txmgr.NewEvmTxmConfig(chain.Config().EVM()) feeCfg := txmgr.NewEvmTxmFeeConfig(chain.Config().EVM().GasEstimator()) + stuckTxDetector := txmgr.NewStuckTxDetector(lggr, ethClient.ConfiguredChainID(), "", assets.NewWei(assets.NewEth(100).ToInt()), chain.Config().EVM().Transactions().AutoPurge(), nil, orm, ethClient) ec := txmgr.NewEvmConfirmer(orm, txmgr.NewEvmTxmClient(ethClient, chain.Config().EVM().NodePool().Errors()), - cfg, feeCfg, chain.Config().EVM().Transactions(), app.GetConfig().Database(), keyStore.Eth(), txBuilder, chain.Logger()) + cfg, feeCfg, chain.Config().EVM().Transactions(), app.GetConfig().Database(), keyStore.Eth(), txBuilder, chain.Logger(), stuckTxDetector) totalNonces := endingNonce - beginningNonce + 1 nonces := make([]evmtypes.Nonce, totalNonces) for i := int64(0); i < totalNonces; i++ { diff --git a/core/config/docs/chains-evm.toml b/core/config/docs/chains-evm.toml index 49360caad21..a4cf1ad7411 100644 --- a/core/config/docs/chains-evm.toml +++ b/core/config/docs/chains-evm.toml @@ -112,6 +112,16 @@ ReaperThreshold = '168h' # Default # ResendAfterThreshold controls how long to wait before re-broadcasting a transaction that has not yet been confirmed. ResendAfterThreshold = '1m' # Default +[EVM.Transactions.AutoPurge] +# Enabled enables or disables automatically purging transactions that have been idenitified as terminally stuck (will never be included on-chain). This feature is only expected to be used by ZK chains. +Enabled = false # Default +# DetectionApiUrl configures the base url of a custom endpoint used to identify terminally stuck transactions. +DetectionApiUrl = 'https://example.api.io' # Example +# Threshold configures the number of blocks a transaction has to remain unconfirmed before it is evaluated for being terminally stuck. This threshold is only applied if there is no custom API to identify stuck transactions provided by the chain. +Threshold = 5 # Example +# MinAttempts configures the minimum number of broadcasted attempts a transaction has to have before it is evaluated further for being terminally stuck. This threshold is only applied if there is no custom API to identify stuck transactions provided by the chain. Ensure the gas estimator configs take more bump attempts before reaching the configured max gas price. +MinAttempts = 3 # Example + [EVM.BalanceMonitor] # Enabled balance monitoring for all keys. Enabled = true # Default diff --git a/core/config/docs/docs_test.go b/core/config/docs/docs_test.go index 1f76eedcc67..fd59edbab6a 100644 --- a/core/config/docs/docs_test.go +++ b/core/config/docs/docs_test.go @@ -87,6 +87,11 @@ func TestDoc(t *testing.T) { docDefaults.ChainWriter.ForwarderAddress = nil docDefaults.NodePool.Errors = evmcfg.ClientErrors{} + // Transactions.AutoPurge configs are only set if the feature is enabled + docDefaults.Transactions.AutoPurge.DetectionApiUrl = nil + docDefaults.Transactions.AutoPurge.Threshold = nil + docDefaults.Transactions.AutoPurge.MinAttempts = nil + assertTOML(t, fallbackDefaults, docDefaults) }) diff --git a/core/services/chainlink/config_test.go b/core/services/chainlink/config_test.go index 8119021b565..2aa1d26c326 100644 --- a/core/services/chainlink/config_test.go +++ b/core/services/chainlink/config_test.go @@ -566,6 +566,9 @@ func TestConfig_Marshal(t *testing.T) { ReaperThreshold: &minute, ResendAfterThreshold: &hour, ForwardersEnabled: ptr(true), + AutoPurge: evmcfg.AutoPurgeConfig{ + Enabled: ptr(false), + }, }, HeadTracker: evmcfg.HeadTracker{ @@ -987,6 +990,9 @@ ReaperInterval = '1m0s' ReaperThreshold = '1m0s' ResendAfterThreshold = '1h0m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true @@ -1216,6 +1222,15 @@ func TestConfig_full(t *testing.T) { got.EVM[c].Nodes[n].Order = ptr(int32(100)) } } + if got.EVM[c].Transactions.AutoPurge.Threshold == nil { + got.EVM[c].Transactions.AutoPurge.Threshold = ptr(uint32(0)) + } + if got.EVM[c].Transactions.AutoPurge.MinAttempts == nil { + got.EVM[c].Transactions.AutoPurge.MinAttempts = ptr(uint32(0)) + } + if got.EVM[c].Transactions.AutoPurge.DetectionApiUrl == nil { + got.EVM[c].Transactions.AutoPurge.DetectionApiUrl = new(commoncfg.URL) + } } cfgtest.AssertFieldsNotNil(t, got) @@ -1242,7 +1257,7 @@ func TestConfig_Validate(t *testing.T) { - LDAP.RunUserGroupCN: invalid value (): LDAP ReadUserGroupCN can not be empty - LDAP.RunUserGroupCN: invalid value (): LDAP RunUserGroupCN can not be empty - LDAP.ReadUserGroupCN: invalid value (): LDAP ReadUserGroupCN can not be empty - - EVM: 8 errors: + - EVM: 9 errors: - 1.ChainID: invalid value (1): duplicate - must be unique - 0.Nodes.1.Name: invalid value (foo): duplicate - must be unique - 3.Nodes.4.WSURL: invalid value (ws://dupe.com): duplicate - must be unique @@ -1260,11 +1275,14 @@ func TestConfig_Validate(t *testing.T) { - WSURL: missing: required for primary nodes - HTTPURL: missing: required for all nodes - 1.HTTPURL: missing: required for all nodes - - 1: 6 errors: + - 1: 9 errors: - ChainType: invalid value (Foo): must not be set with this chain id - Nodes: missing: must have at least one node - - ChainType: invalid value (Foo): must be one of arbitrum, celo, gnosis, kroma, metis, optimismBedrock, scroll, wemix, xlayer, zksync or omitted + - ChainType: invalid value (Foo): must be one of arbitrum, celo, gnosis, kroma, metis, optimismBedrock, scroll, wemix, xlayer, zkevm, zksync or omitted - HeadTracker.HistoryDepth: invalid value (30): must be equal to or greater than FinalityDepth + - GasEstimator.BumpThreshold: invalid value (0): cannot be 0 if auto-purge feature is enabled for Foo + - Transactions.AutoPurge.Threshold: missing: needs to be set if auto-purge feature is enabled for Foo + - Transactions.AutoPurge.MinAttempts: missing: needs to be set if auto-purge feature is enabled for Foo - GasEstimator: 2 errors: - FeeCapDefault: invalid value (101 wei): must be equal to PriceMax (99 wei) since you are using FixedPrice estimation with gas bumping disabled in EIP1559 mode - PriceMax will be used as the FeeCap for transactions instead of FeeCapDefault - PriceMax: invalid value (1 gwei): must be greater than or equal to PriceDefault @@ -1272,7 +1290,7 @@ func TestConfig_Validate(t *testing.T) { - 2: 5 errors: - ChainType: invalid value (Arbitrum): only "optimismBedrock" can be used with this chain id - Nodes: missing: must have at least one node - - ChainType: invalid value (Arbitrum): must be one of arbitrum, celo, gnosis, kroma, metis, optimismBedrock, scroll, wemix, xlayer, zksync or omitted + - ChainType: invalid value (Arbitrum): must be one of arbitrum, celo, gnosis, kroma, metis, optimismBedrock, scroll, wemix, xlayer, zkevm, zksync or omitted - FinalityDepth: invalid value (0): must be greater than or equal to 1 - MinIncomingConfirmations: invalid value (0): must be greater than or equal to 1 - 3.Nodes: 5 errors: @@ -1293,6 +1311,7 @@ func TestConfig_Validate(t *testing.T) { - 4: 2 errors: - ChainID: missing: required for all chains - Nodes: missing: must have at least one node + - 5.Transactions.AutoPurge.DetectionApiUrl: invalid value (): must be set for scroll - Cosmos: 5 errors: - 1.ChainID: invalid value (Malaga-420): duplicate - must be unique - 0.Nodes.1.Name: invalid value (test): duplicate - must be unique diff --git a/core/services/chainlink/testdata/config-full.toml b/core/services/chainlink/testdata/config-full.toml index b199ae530f5..356a6d69930 100644 --- a/core/services/chainlink/testdata/config-full.toml +++ b/core/services/chainlink/testdata/config-full.toml @@ -291,6 +291,9 @@ ReaperInterval = '1m0s' ReaperThreshold = '1m0s' ResendAfterThreshold = '1h0m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/core/services/chainlink/testdata/config-invalid.toml b/core/services/chainlink/testdata/config-invalid.toml index 7d1ed17c3c2..b53f4b1712d 100644 --- a/core/services/chainlink/testdata/config-invalid.toml +++ b/core/services/chainlink/testdata/config-invalid.toml @@ -54,6 +54,9 @@ ChainID = '1' ChainType = 'Foo' FinalityDepth = 32 +[EVM.Transactions.AutoPurge] +Enabled = true + [EVM.GasEstimator] Mode = 'FixedPrice' BumpThreshold = 0 @@ -99,6 +102,19 @@ WSURL = 'ws://dupe.com' [[EVM]] +[[EVM]] +ChainID = '534352' +ChainType = 'scroll' + +[EVM.Transactions.AutoPurge] +Enabled = true +DetectionApiUrl = '' + +[[EVM.Nodes]] +Name = 'scroll node' +WSURL = 'ws://foo.bar' +HTTPURl = 'http://foo.bar' + [[Cosmos]] ChainID = 'Malaga-420' diff --git a/core/services/chainlink/testdata/config-multi-chain-effective.toml b/core/services/chainlink/testdata/config-multi-chain-effective.toml index 7aa3bb50b35..25d62801455 100644 --- a/core/services/chainlink/testdata/config-multi-chain-effective.toml +++ b/core/services/chainlink/testdata/config-multi-chain-effective.toml @@ -278,6 +278,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true @@ -369,6 +372,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true @@ -454,6 +460,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/core/services/ocr/contract_tracker.go b/core/services/ocr/contract_tracker.go index 94ad1237e90..34852bbe74b 100644 --- a/core/services/ocr/contract_tracker.go +++ b/core/services/ocr/contract_tracker.go @@ -400,7 +400,7 @@ func (t *OCRContractTracker) LatestBlockHeight(ctx context.Context) (blockheight // care about the block height; we have no way of getting the L1 block // height anyway return 0, nil - case "", config.ChainArbitrum, config.ChainCelo, config.ChainGnosis, config.ChainKroma, config.ChainOptimismBedrock, config.ChainScroll, config.ChainWeMix, config.ChainXLayer, config.ChainZkSync: + case "", config.ChainArbitrum, config.ChainCelo, config.ChainGnosis, config.ChainKroma, config.ChainOptimismBedrock, config.ChainScroll, config.ChainWeMix, config.ChainXLayer, config.ChainZkEvm, config.ChainZkSync: // continue } latestBlockHeight := t.getLatestBlockHeight() diff --git a/core/services/ocrcommon/block_translator.go b/core/services/ocrcommon/block_translator.go index 06fd9941992..7bce661e692 100644 --- a/core/services/ocrcommon/block_translator.go +++ b/core/services/ocrcommon/block_translator.go @@ -21,7 +21,7 @@ func NewBlockTranslator(cfg Config, client evmclient.Client, lggr logger.Logger) switch cfg.ChainType() { case config.ChainArbitrum: return NewArbitrumBlockTranslator(client, lggr) - case "", config.ChainCelo, config.ChainGnosis, config.ChainKroma, config.ChainMetis, config.ChainOptimismBedrock, config.ChainScroll, config.ChainWeMix, config.ChainXLayer, config.ChainZkSync: + case "", config.ChainCelo, config.ChainGnosis, config.ChainKroma, config.ChainMetis, config.ChainOptimismBedrock, config.ChainScroll, config.ChainWeMix, config.ChainXLayer, config.ChainZkEvm, config.ChainZkSync: fallthrough default: return &l1BlockTranslator{} diff --git a/core/store/migrate/migrations/0239_add_purge_column_tx_attempts.sql b/core/store/migrate/migrations/0239_add_purge_column_tx_attempts.sql new file mode 100644 index 00000000000..af23032d02a --- /dev/null +++ b/core/store/migrate/migrations/0239_add_purge_column_tx_attempts.sql @@ -0,0 +1,32 @@ +-- +goose Up +ALTER TABLE evm.tx_attempts ADD COLUMN is_purge_attempt boolean NOT NULL DEFAULT false; +ALTER TABLE evm.txes DROP CONSTRAINT chk_eth_txes_fsm; +ALTER TABLE evm.txes ADD CONSTRAINT chk_eth_txes_fsm CHECK ( + state = 'unstarted'::eth_txes_state AND nonce IS NULL AND error IS NULL AND broadcast_at IS NULL AND initial_broadcast_at IS NULL + OR + state = 'in_progress'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NULL AND initial_broadcast_at IS NULL + OR + state = 'fatal_error'::eth_txes_state AND error IS NOT NULL + OR + state = 'unconfirmed'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL + OR + state = 'confirmed'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL + OR + state = 'confirmed_missing_receipt'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL +) NOT VALID; +-- +goose Down +ALTER TABLE evm.tx_attempts DROP COLUMN is_purge_attempt; +ALTER TABLE evm.txes DROP CONSTRAINT chk_eth_txes_fsm; +ALTER TABLE evm.txes ADD CONSTRAINT chk_eth_txes_fsm CHECK ( + state = 'unstarted'::eth_txes_state AND nonce IS NULL AND error IS NULL AND broadcast_at IS NULL AND initial_broadcast_at IS NULL + OR + state = 'in_progress'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NULL AND initial_broadcast_at IS NULL + OR + state = 'fatal_error'::eth_txes_state AND nonce IS NULL AND error IS NOT NULL + OR + state = 'unconfirmed'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL + OR + state = 'confirmed'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL + OR + state = 'confirmed_missing_receipt'::eth_txes_state AND nonce IS NOT NULL AND error IS NULL AND broadcast_at IS NOT NULL AND initial_broadcast_at IS NOT NULL +) NOT VALID; \ No newline at end of file diff --git a/core/web/resolver/testdata/config-full.toml b/core/web/resolver/testdata/config-full.toml index 75fad4d2fc9..2b6d728df95 100644 --- a/core/web/resolver/testdata/config-full.toml +++ b/core/web/resolver/testdata/config-full.toml @@ -291,6 +291,9 @@ ReaperInterval = '1m0s' ReaperThreshold = '1m0s' ResendAfterThreshold = '1h0m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/core/web/resolver/testdata/config-multi-chain-effective.toml b/core/web/resolver/testdata/config-multi-chain-effective.toml index 7aa3bb50b35..25d62801455 100644 --- a/core/web/resolver/testdata/config-multi-chain-effective.toml +++ b/core/web/resolver/testdata/config-multi-chain-effective.toml @@ -278,6 +278,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true @@ -369,6 +372,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true @@ -454,6 +460,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/docs/CONFIG.md b/docs/CONFIG.md index a0e2957cd72..c7b34dea7f2 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -1758,6 +1758,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -1843,6 +1846,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -1928,6 +1934,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2013,6 +2022,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2099,6 +2111,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2184,6 +2199,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2269,6 +2287,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2355,6 +2376,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2440,6 +2464,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2524,6 +2551,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2608,6 +2638,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2693,6 +2726,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2779,6 +2815,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2864,6 +2903,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -2949,6 +2991,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3034,6 +3079,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3119,6 +3167,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3204,6 +3255,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3289,6 +3343,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3374,6 +3431,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3459,6 +3519,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3544,6 +3607,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3630,6 +3696,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3715,6 +3784,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3799,6 +3871,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3884,6 +3959,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -3946,6 +4024,7 @@ GasLimit = 5400000 AutoCreateKey = true BlockBackfillDepth = 10 BlockBackfillSkip = false +ChainType = 'zkevm' FinalityDepth = 500 FinalityTagEnabled = false LogBackfillBatchSize = 1000 @@ -3968,6 +4047,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4053,6 +4135,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4138,6 +4223,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4222,6 +4310,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '0s' ResendAfterThreshold = '0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4284,6 +4375,7 @@ GasLimit = 5400000 AutoCreateKey = true BlockBackfillDepth = 10 BlockBackfillSkip = false +ChainType = 'zkevm' FinalityDepth = 500 FinalityTagEnabled = false LogBackfillBatchSize = 1000 @@ -4306,6 +4398,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4391,6 +4486,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4453,6 +4551,7 @@ GasLimit = 5400000 AutoCreateKey = true BlockBackfillDepth = 10 BlockBackfillSkip = false +ChainType = 'zkevm' FinalityDepth = 500 FinalityTagEnabled = false LogBackfillBatchSize = 1000 @@ -4475,6 +4574,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4560,6 +4662,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4644,6 +4749,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4729,6 +4837,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4814,6 +4925,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4900,6 +5014,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -4985,6 +5102,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5070,6 +5190,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5155,6 +5278,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5240,6 +5366,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5324,6 +5453,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5408,6 +5540,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5492,6 +5627,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '3m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5577,6 +5715,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5662,6 +5803,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5746,6 +5890,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5831,6 +5978,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -5916,6 +6066,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6002,6 +6155,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6088,6 +6244,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6173,6 +6332,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6258,6 +6420,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6343,6 +6508,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6428,6 +6596,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6513,6 +6684,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '30s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6598,6 +6772,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6683,6 +6860,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[Transactions.AutoPurge] +Enabled = false + [BalanceMonitor] Enabled = true @@ -6971,6 +7151,40 @@ ResendAfterThreshold = '1m' # Default ``` ResendAfterThreshold controls how long to wait before re-broadcasting a transaction that has not yet been confirmed. +## EVM.Transactions.AutoPurge +```toml +[EVM.Transactions.AutoPurge] +Enabled = false # Default +DetectionApiUrl = 'https://example.api.io' # Example +Threshold = 5 # Example +MinAttempts = 3 # Example +``` + + +### Enabled +```toml +Enabled = false # Default +``` +Enabled enables or disables automatically purging transactions that have been idenitified as terminally stuck (will never be included on-chain). This feature is only expected to be used by ZK chains. + +### DetectionApiUrl +```toml +DetectionApiUrl = 'https://example.api.io' # Example +``` +DetectionApiUrl configures the base url of a custom endpoint used to identify terminally stuck transactions. + +### Threshold +```toml +Threshold = 5 # Example +``` +Threshold configures the number of blocks a transaction has to remain unconfirmed before it is evaluated for being terminally stuck. This threshold is only applied if there is no custom API to identify stuck transactions provided by the chain. + +### MinAttempts +```toml +MinAttempts = 3 # Example +``` +MinAttempts configures the minimum number of broadcasted attempts a transaction has to have before it is evaluated further for being terminally stuck. This threshold is only applied if there is no custom API to identify stuck transactions provided by the chain. Ensure the gas estimator configs take more bump attempts before reaching the configured max gas price. + ## EVM.BalanceMonitor ```toml [EVM.BalanceMonitor] diff --git a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar index feaf546f022..95366d92f54 100644 --- a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar @@ -334,6 +334,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar index b37fed41150..02b3eaba50d 100644 --- a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar @@ -334,6 +334,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/testdata/scripts/node/validate/disk-based-logging.txtar b/testdata/scripts/node/validate/disk-based-logging.txtar index 6ae02ab38f4..d83e54018b5 100644 --- a/testdata/scripts/node/validate/disk-based-logging.txtar +++ b/testdata/scripts/node/validate/disk-based-logging.txtar @@ -334,6 +334,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/testdata/scripts/node/validate/invalid.txtar b/testdata/scripts/node/validate/invalid.txtar index df0118bbbbf..065be4222de 100644 --- a/testdata/scripts/node/validate/invalid.txtar +++ b/testdata/scripts/node/validate/invalid.txtar @@ -324,6 +324,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/testdata/scripts/node/validate/valid.txtar b/testdata/scripts/node/validate/valid.txtar index edb07fd5e4f..8730cb4b0d5 100644 --- a/testdata/scripts/node/validate/valid.txtar +++ b/testdata/scripts/node/validate/valid.txtar @@ -331,6 +331,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true diff --git a/testdata/scripts/node/validate/warnings.txtar b/testdata/scripts/node/validate/warnings.txtar index 54de3227a9e..7d45a97cc40 100644 --- a/testdata/scripts/node/validate/warnings.txtar +++ b/testdata/scripts/node/validate/warnings.txtar @@ -330,6 +330,9 @@ ReaperInterval = '1h0m0s' ReaperThreshold = '168h0m0s' ResendAfterThreshold = '1m0s' +[EVM.Transactions.AutoPurge] +Enabled = false + [EVM.BalanceMonitor] Enabled = true