Merge upstream txpool metrics

core/txpool/invalid.go

@@ -5,7 +5,6 @@ import (
 )
 
 const (
-	AlreadyKnown            = "AlreadyKnown"
 	TypeNotSupportDeposit   = "TypeNotSupportDeposit"
 	TypeNotSupport1559      = "TypeNotSupport1559"
 	TypeNotSupport2718      = "TypeNotSupport2718"
@@ -23,22 +22,21 @@ const (
 	InsufficientFunds       = "InsufficientFunds"
 	Overdraft               = "Overdraft"
 	IntrinsicGas            = "IntrinsicGas"
-	Throttle                = "Throttle"
-	Overflow                = "Overflow"
 	FutureReplacePending    = "FutureReplacePending"
-	ReplaceUnderpriced      = "ReplaceUnderpriced"
-	QueuedDiscard           = "QueueDiscard"
 	GasUnitOverflow         = "GasUnitOverflow"
 )
 
+func Meter(err string) metrics.Meter {
+	return meter(err)
+}
+
 func meter(err string) metrics.Meter {
 	return metrics.GetOrRegisterMeter("txpool/invalid/"+err, nil)
 }
 
 func init() {
 	// init the metrics
 	for _, err := range []string{
-		AlreadyKnown,
 		TypeNotSupportDeposit,
 		TypeNotSupport1559,
 		TypeNotSupport2718,
@@ -56,11 +54,7 @@ func init() {
 		InsufficientFunds,
 		Overdraft,
 		IntrinsicGas,
-		Throttle,
-		Overflow,
 		FutureReplacePending,
-		ReplaceUnderpriced,
-		QueuedDiscard,
 		GasUnitOverflow,
 	} {
 		meter(err).Mark(0)

core/txpool/legacypool/legacypool.go

@@ -829,6 +829,7 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 					pool.priced.Put(dropTx, false)
 				}
 				log.Trace("Discarding future transaction replacing pending tx", "hash", hash)
+				txpool.Meter(txpool.FutureReplacePending).Mark(1)
 				return false, txpool.ErrFutureReplacePending
 			}
 		}
@@ -940,6 +941,7 @@ func (pool *LegacyPool) enqueueTx(hash common.Hash, tx *types.Transaction, local
 	// If the transaction isn't in lookup set but it's expected to be there,
 	// show the error log.
 	if pool.all.Get(hash) == nil && !addAll {
+		txpool.Meter(txpool.MissingTransaction).Mark(1)
 		log.Error("Missing transaction in lookup set, please report the issue", "hash", hash)
 	}
 	if addAll {

core/txpool/validation.go

@@ -129,9 +129,11 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
 	// the transaction metadata
 	intrGas, err := core.IntrinsicGas(tx.Data(), tx.AccessList(), tx.To() == nil, true, opts.Config.IsIstanbul(head.Number), opts.Config.IsShanghai(head.Number, head.Time))
 	if err != nil {
+		meter(GasUnitOverflow).Mark(1)
 		return err
 	}
 	if tx.Gas() < intrGas {
+		meter(IntrinsicGas).Mark(1)
 		return fmt.Errorf("%w: needed %v, allowed %v", core.ErrIntrinsicGas, intrGas, tx.Gas())
 	}
 	// Ensure the gasprice is high enough to cover the requirement of the calling
@@ -271,11 +273,13 @@ func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, op
 		bump := new(big.Int).Sub(cost, prev)
 		need := new(big.Int).Add(spent, bump)
 		if balance.Cmp(need) < 0 {
+			meter(Overdraft).Mark(1)
 			return fmt.Errorf("%w: balance %v, queued cost %v, tx bumped %v, overshot %v", core.ErrInsufficientFunds, balance, spent, bump, new(big.Int).Sub(need, balance))
 		}
 	} else {
 		need := new(big.Int).Add(spent, cost)
 		if balance.Cmp(need) < 0 {
+			meter(Overdraft).Mark(1)
 			return fmt.Errorf("%w: balance %v, queued cost %v, tx cost %v, overshot %v", core.ErrInsufficientFunds, balance, spent, cost, new(big.Int).Sub(need, balance))
 		}
 		// Transaction takes a new nonce value out of the pool. Ensure it doesn't

eth/handler_eth_test.go

@@ -461,14 +461,14 @@ func TestTransactionPendingReannounce(t *testing.T) {
 
 	sink := newTestHandler()
 	defer sink.close()
-	sink.handler.acceptTxs = 1 // mark synced to accept transactions
+	sink.handler.synced.Store(true) // mark synced to accept transactions
 
 	sourcePipe, sinkPipe := p2p.MsgPipe()
 	defer sourcePipe.Close()
 	defer sinkPipe.Close()
 
-	sourcePeer := eth.NewPeer(eth.ETH66, p2p.NewPeer(enode.ID{0}, "", nil), sourcePipe, source.txpool)
-	sinkPeer := eth.NewPeer(eth.ETH66, p2p.NewPeer(enode.ID{0}, "", nil), sinkPipe, sink.txpool)
+	sourcePeer := eth.NewPeer(eth.ETH68, p2p.NewPeer(enode.ID{0}, "", nil), sourcePipe, source.txpool)
+	sinkPeer := eth.NewPeer(eth.ETH68, p2p.NewPeer(enode.ID{0}, "", nil), sinkPipe, sink.txpool)
 	defer sourcePeer.Close()
 	defer sinkPeer.Close()
 
@@ -481,7 +481,7 @@ func TestTransactionPendingReannounce(t *testing.T) {
 
 	// Subscribe transaction pools
 	txCh := make(chan core.NewTxsEvent, 1024)
-	sub := sink.txpool.SubscribeNewTxsEvent(txCh)
+	sub := sink.txpool.SubscribeTransactions(txCh, false)
 	defer sub.Unsubscribe()
 
 	txs := make([]*types.Transaction, 64)