fix: txpool metrics and unit test (bnb-chain#59)

Co-authored-by: andyzhang2023 <andyzhang2023@gmail.com>
Co-authored-by: bnoieh <135800952+bnoieh@users.noreply.github.com>

core/txpool/invalid.go

@@ -5,7 +5,6 @@ import (
 )
 
 const (
- AlreadyKnown = "AlreadyKnown"
  TypeNotSupportDeposit = "TypeNotSupportDeposit"
  TypeNotSupport1559 = "TypeNotSupport1559"
  TypeNotSupport2718 = "TypeNotSupport2718"
@@ -23,22 +22,17 @@ 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 {
+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,13 +50,9 @@ func init() {
  InsufficientFunds,
  Overdraft,
  IntrinsicGas,
- Throttle,
- Overflow,
  FutureReplacePending,
- ReplaceUnderpriced,
- QueuedDiscard,
  GasUnitOverflow,
  } {
- meter(err).Mark(0)
+ 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

@@ -67,7 +67,7 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
  // This is for spam protection, not consensus,
  // as the external engine-API user authenticates deposits.
  if tx.Type() == types.DepositTxType {
- meter(TypeNotSupportDeposit).Mark(1)
+ Meter(TypeNotSupportDeposit).Mark(1)
  return core.ErrTxTypeNotSupported
  }
  // Ensure transactions not implemented by the calling pool are rejected
@@ -81,25 +81,25 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
  }
  // Ensure only transactions that have been enabled are accepted
  if !opts.Config.IsBerlin(head.Number) && tx.Type() != types.LegacyTxType {
- meter(TypeNotSupport2718).Mark(1)
+ Meter(TypeNotSupport2718).Mark(1)
  return fmt.Errorf("%w: type %d rejected, pool not yet in Berlin", core.ErrTxTypeNotSupported, tx.Type())
  }
  if !opts.Config.IsLondon(head.Number) && tx.Type() == types.DynamicFeeTxType {
- meter(TypeNotSupport1559).Mark(1)
+ Meter(TypeNotSupport1559).Mark(1)
  return fmt.Errorf("%w: type %d rejected, pool not yet in London", core.ErrTxTypeNotSupported, tx.Type())
  }
  if !opts.Config.IsCancun(head.Number, head.Time) && tx.Type() == types.BlobTxType {
  return fmt.Errorf("%w: type %d rejected, pool not yet in Cancun", core.ErrTxTypeNotSupported, tx.Type())
  }
  // Check whether the init code size has been exceeded
  if opts.Config.IsShanghai(head.Number, head.Time) && tx.To() == nil && len(tx.Data()) > params.MaxInitCodeSize {
- meter(MaxInitCodeSizeExceeded).Mark(1)
+ Meter(MaxInitCodeSizeExceeded).Mark(1)
  return fmt.Errorf("%w: code size %v, limit %v", core.ErrMaxInitCodeSizeExceeded, len(tx.Data()), params.MaxInitCodeSize)
  }
  // Transactions can't be negative. This may never happen using RLP decoded
  // transactions but may occur for transactions created using the RPC.
  if tx.Value().Sign() < 0 {
- meter(NegativeValue).Mark(1)
+ Meter(NegativeValue).Mark(1)
  return ErrNegativeValue
  }
  // Ensure the transaction doesn't exceed the current block limit gas
@@ -108,36 +108,38 @@ func ValidateTransaction(tx *types.Transaction, head *types.Header, signer types
  }
  // Sanity check for extremely large numbers (supported by RLP or RPC)
  if tx.GasFeeCap().BitLen() > 256 {
- meter(FeeCapVeryHigh).Mark(1)
+ Meter(FeeCapVeryHigh).Mark(1)
  return core.ErrFeeCapVeryHigh
  }
  if tx.GasTipCap().BitLen() > 256 {
- meter(TipVeryHigh).Mark(1)
+ Meter(TipVeryHigh).Mark(1)
  return core.ErrTipVeryHigh
  }
  // Ensure gasFeeCap is greater than or equal to gasTipCap
  if tx.GasFeeCapIntCmp(tx.GasTipCap()) < 0 {
- meter(TipAboveFeeCap).Mark(1)
+ Meter(TipAboveFeeCap).Mark(1)
  return core.ErrTipAboveFeeCap
  }
  // Make sure the transaction is signed properly
  if _, err := types.Sender(signer, tx); err != nil {
- meter(InvalidSender).Mark(1)
+ Meter(InvalidSender).Mark(1)
  return ErrInvalidSender
  }
  // Ensure the transaction has more gas than the bare minimum needed to cover
  // 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
  // pool and/or block producer
  if tx.GasTipCapIntCmp(opts.MinTip) < 0 {
- meter(Underpriced).Mark(1)
+ Meter(Underpriced).Mark(1)
  return fmt.Errorf("%w: tip needed %v, tip permitted %v", ErrUnderpriced, opts.MinTip, tx.GasTipCap())
  }
  // Ensure blob transactions have valid commitments
@@ -240,7 +242,7 @@ func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, op
  }
  next := opts.State.GetNonce(from)
  if next > tx.Nonce() {
- meter(NonceTooLow).Mark(1)
+ Meter(NonceTooLow).Mark(1)
  return fmt.Errorf("%w: next nonce %v, tx nonce %v", core.ErrNonceTooLow, next, tx.Nonce())
  }
  // Ensure the transaction doesn't produce a nonce gap in pools that do not
@@ -261,7 +263,7 @@ func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, op
  }
  }
  if balance.Cmp(cost) < 0 {
- meter(InsufficientFunds).Mark(1)
+ Meter(InsufficientFunds).Mark(1)
  return fmt.Errorf("%w: balance %v, tx cost %v, overshot %v", core.ErrInsufficientFunds, balance, cost, new(big.Int).Sub(cost, balance))
  }
  // Ensure the transactor has enough funds to cover for replacements or nonce
@@ -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)