txpool: Reorganize Filter Checks (#572)

Author: axelKingsley

core/txpool/blobpool/blobpool.go

@@ -348,6 +348,10 @@ func (p *BlobPool) Filter(tx *types.Transaction) bool {
 	return tx.Type() == types.BlobTxType
 }
 
+func (p *BlobPool) SetIngressFilters(filters []txpool.IngressFilter) {
+	// No-op, ingress filters are not supported in the blob pool
+}
+
 // Init sets the gas price needed to keep a transaction in the pool and the chain
 // head to allow balance / nonce checks. The transaction journal will be loaded
 // from disk and filtered based on the provided starting settings.

core/txpool/legacypool/legacypool.go

@@ -18,6 +18,7 @@
 package legacypool
 
 import (
+	"context"
 	"errors"
 	"math"
 	"math/big"
@@ -261,6 +262,10 @@ type LegacyPool struct {
 	changesSinceReorg int // A counter for how many drops we've performed in-between reorg.
 
 	rollupCostFn txpool.RollupCostFunc // Additional rollup cost function, optional field, may be nil.
+
+	ingressFilters []txpool.IngressFilter // Filters to apply to incoming transactions
+	filterCtx      context.Context        // Filters may use this context with external resources
+	filterCancel   context.CancelFunc     // Cancel function for the filter context
 }
 
 type txpoolResetRequest struct {
@@ -397,6 +402,10 @@ func (pool *LegacyPool) loop() {
 
 // Close terminates the transaction pool.
 func (pool *LegacyPool) Close() error {
+	// Cancel the filter context if it exists
+	if pool.filterCancel != nil {
+		pool.filterCancel()
+	}
 	// Terminate the pool reorger and return
 	close(pool.reorgShutdownCh)
 	pool.wg.Wait()
@@ -477,6 +486,13 @@ func (pool *LegacyPool) stats() (int, int) {
 	return pending, queued
 }
 
+func (pool *LegacyPool) SetIngressFilters(filters []txpool.IngressFilter) {
+	pool.mu.Lock()
+	defer pool.mu.Unlock()
+	pool.filterCtx, pool.filterCancel = context.WithCancel(context.Background())
+	pool.ingressFilters = filters
+}
+
 // Content retrieves the data content of the transaction pool, returning all the
 // pending as well as queued transactions, grouped by account and sorted by nonce.
 func (pool *LegacyPool) Content() (map[common.Address][]*types.Transaction, map[common.Address][]*types.Transaction) {
@@ -987,6 +1003,20 @@ func (pool *LegacyPool) Add(txs []*types.Transaction, sync bool) []error {
 			invalidTxMeter.Mark(1)
 			continue
 		}
+		// Exclude transactions which fail the ingress filters
+		filtered := false
+		for _, filter := range pool.ingressFilters {
+			if !filter.FilterTx(pool.filterCtx, tx) {
+				errs[i] = core.ErrTxFilteredOut
+				log.Trace("Discarding filtered transaction", "hash", tx.Hash())
+				invalidTxMeter.Mark(1)
+				filtered = true
+				break
+			}
+		}
+		if filtered {
+			continue
+		}
 		// Accumulate all unknown transactions for deeper processing
 		news = append(news, tx)
 	}

core/txpool/subpool.go

@@ -111,6 +111,8 @@ type SubPool interface {
 	// one another.
 	Init(gasTip uint64, head *types.Header, reserve AddressReserver) error
 
+	SetIngressFilters([]IngressFilter)
+
 	// Close terminates any background processing threads and releases any held
 	// resources.
 	Close() error

core/txpool/txpool.go

@@ -17,7 +17,6 @@
 package txpool
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"math/big"
@@ -77,32 +76,22 @@ type TxPool struct {
 	term chan struct{}           // Termination channel to detect a closed pool
 
 	sync chan chan error // Testing / simulator channel to block until internal reset is done
-
-	ingressFilters []IngressFilter // List of filters to apply to incoming transactions
-
-	filterCtx    context.Context    // Filters may use external resources
-	filterCancel context.CancelFunc // Filter calls are cancelled on shutdown
 }
 
 // New creates a new transaction pool to gather, sort and filter inbound
 // transactions from the network.
-func New(gasTip uint64, chain BlockChain, subpools []SubPool, poolFilters []IngressFilter) (*TxPool, error) {
+func New(gasTip uint64, chain BlockChain, subpools []SubPool, ingressFilters []IngressFilter) (*TxPool, error) {
 	// Retrieve the current head so that all subpools and this main coordinator
 	// pool will have the same starting state, even if the chain moves forward
 	// during initialization.
 	head := chain.CurrentBlock()
 
-	filterCtx, filterCancel := context.WithCancel(context.Background())
-
 	pool := &TxPool{
-		subpools:       subpools,
-		reservations:   make(map[common.Address]SubPool),
-		quit:           make(chan chan error),
-		term:           make(chan struct{}),
-		sync:           make(chan chan error),
-		ingressFilters: poolFilters,
-		filterCtx:      filterCtx,
-		filterCancel:   filterCancel,
+		subpools:     subpools,
+		reservations: make(map[common.Address]SubPool),
+		quit:         make(chan chan error),
+		term:         make(chan struct{}),
+		sync:         make(chan chan error),
 	}
 	for i, subpool := range subpools {
 		if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil {
@@ -111,6 +100,8 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool, poolFilters []Ingr
 			}
 			return nil, err
 		}
+		// Set the ingress filters for the subpool
+		subpool.SetIngressFilters(ingressFilters)
 	}
 	go pool.loop(head, chain)
 	return pool, nil
@@ -166,8 +157,6 @@ func (p *TxPool) reserver(id int, subpool SubPool) AddressReserver {
 func (p *TxPool) Close() error {
 	var errs []error
 
-	p.filterCancel() // Cancel filter work, these in-flight txs will be not be allowed through before shutdown
-
 	// Terminate the reset loop and wait for it to finish
 	errc := make(chan error)
 	p.quit <- errc
@@ -350,23 +339,11 @@ func (p *TxPool) Add(txs []*types.Transaction, sync bool) []error {
 	// so we can piece back the returned errors into the original order.
 	txsets := make([][]*types.Transaction, len(p.subpools))
 	splits := make([]int, len(txs))
-	filtered_out := make([]bool, len(txs))
 
 	for i, tx := range txs {
 		// Mark this transaction belonging to no-subpool
 		splits[i] = -1
 
-		// Filter the transaction through the ingress filters
-		for _, f := range p.ingressFilters {
-			if !f.FilterTx(p.filterCtx, tx) {
-				filtered_out[i] = true
-			}
-		}
-		// if the transaction is filtered out, don't add it to any subpool
-		if filtered_out[i] {
-			continue
-		}
-
 		// Try to find a subpool that accepts the transaction
 		for j, subpool := range p.subpools {
 			if subpool.Filter(tx) {
@@ -384,11 +361,6 @@ func (p *TxPool) Add(txs []*types.Transaction, sync bool) []error {
 	}
 	errs := make([]error, len(txs))
 	for i, split := range splits {
-		// If the transaction was filtered out, mark it as such
-		if filtered_out[i] {
-			errs[i] = core.ErrTxFilteredOut
-			continue
-		}
 		// If the transaction was rejected by all subpools, mark it unsupported
 		if split == -1 {
 			errs[i] = fmt.Errorf("%w: received type %d", core.ErrTxTypeNotSupported, txs[i].Type())