eth/fetcher: move check for chain tx inside fetcher

Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>

eth/fetcher/tx_fetcher.go

@@ -27,6 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
@@ -71,6 +72,11 @@ const (
 
 	// addTxsBatchSize it the max number of transactions to add in a single batch from a peer.
 	addTxsBatchSize = 128
+
+	// txOnChainCacheLimit is number of on-chain transactions to keep in a cache to avoid
+	// re-fetching them soon after they are mined.
+	// Approx 1MB for 30 minutes of transactions at 18 tps
+	txOnChainCacheLimit = 32768
 )
 
 var (
@@ -152,6 +158,9 @@ type TxFetcher struct {
 	txSeq       uint64                             // Unique transaction sequence number
 	underpriced *lru.Cache[common.Hash, time.Time] // Transactions discarded as too cheap (don't re-fetch)
 
+	chain          *core.BlockChain                  // Blockchain interface for on-chain checks
+	txOnChainCache *lru.Cache[common.Hash, struct{}] // Cache to avoid fetching once the tx gets on chain
+
 	// Stage 1: Waiting lists for newly discovered transactions that might be
 	// broadcast without needing explicit request/reply round trips.
 	waitlist  map[common.Hash]map[string]struct{}           // Transactions waiting for an potential broadcast
@@ -184,36 +193,40 @@ type TxFetcher struct {
 
 // NewTxFetcher creates a transaction fetcher to retrieve transaction
 // based on hash announcements.
-func NewTxFetcher(validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher {
+// Chain can be nil to disable on-chain checks.
-	return NewTxFetcherForTests(validateMeta, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil)
+func NewTxFetcher(chain *core.BlockChain, validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher {
+	return NewTxFetcherForTests(chain, validateMeta, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil)
 }
 
 // NewTxFetcherForTests is a testing method to mock out the realtime clock with
 // a simulated version and the internal randomness with a deterministic one.
+// Chain can be nil to disable on-chain checks.
 func NewTxFetcherForTests(
-	validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string),
+	chain *core.BlockChain, validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string),
 	clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher {
 	return &TxFetcher{
-		notify:       make(chan *txAnnounce),
+		notify:         make(chan *txAnnounce),
-		cleanup:      make(chan *txDelivery),
+		cleanup:        make(chan *txDelivery),
-		drop:         make(chan *txDrop),
+		drop:           make(chan *txDrop),
-		quit:         make(chan struct{}),
+		quit:           make(chan struct{}),
-		waitlist:     make(map[common.Hash]map[string]struct{}),
+		waitlist:       make(map[common.Hash]map[string]struct{}),
-		waittime:     make(map[common.Hash]mclock.AbsTime),
+		waittime:       make(map[common.Hash]mclock.AbsTime),
-		waitslots:    make(map[string]map[common.Hash]*txMetadataWithSeq),
+		waitslots:      make(map[string]map[common.Hash]*txMetadataWithSeq),
-		announces:    make(map[string]map[common.Hash]*txMetadataWithSeq),
+		announces:      make(map[string]map[common.Hash]*txMetadataWithSeq),
-		announced:    make(map[common.Hash]map[string]struct{}),
+		announced:      make(map[common.Hash]map[string]struct{}),
-		fetching:     make(map[common.Hash]string),
+		fetching:       make(map[common.Hash]string),
-		requests:     make(map[string]*txRequest),
+		requests:       make(map[string]*txRequest),
-		alternates:   make(map[common.Hash]map[string]struct{}),
+		alternates:     make(map[common.Hash]map[string]struct{}),
-		underpriced:  lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
+		underpriced:    lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
-		validateMeta: validateMeta,
+		txOnChainCache: lru.NewCache[common.Hash, struct{}](txOnChainCacheLimit),
-		addTxs:       addTxs,
+		chain:          chain,
-		fetchTxs:     fetchTxs,
+		validateMeta:   validateMeta,
-		dropPeer:     dropPeer,
+		addTxs:         addTxs,
-		clock:        clock,
+		fetchTxs:       fetchTxs,
-		realTime:     realTime,
+		dropPeer:       dropPeer,
-		rand:         rand,
+		clock:          clock,
+		realTime:       realTime,
+		rand:           rand,
 	}
 }
 
@@ -245,6 +258,12 @@ func (f *TxFetcher) Notify(peer string, types []byte, sizes []uint32, hashes []c
 			continue
 		}
 
+		// check on chain as well (no need to check limbo separately, as chain checks limbo too)
+		if _, exist := f.txOnChainCache.Get(hash); exist {
+			duplicate++ //TODO
+			continue
+		}
+
 		if f.isKnownUnderpriced(hash) {
 			underpriced++
 			continue
@@ -412,7 +431,29 @@ func (f *TxFetcher) loop() {
 
 		waitTrigger    = make(chan struct{}, 1)
 		timeoutTrigger = make(chan struct{}, 1)
+
+		oldHead *types.Header
 	)
+
+	//Subscribe to chain events to know when transactions are added to chain
+	var headEventCh chan core.ChainEvent
+	var subErr <-chan error
+	if f.chain != nil {
+		headEventCh = make(chan core.ChainEvent, 10)
+		sub := f.chain.SubscribeChainEvent(headEventCh)
+		subErr = sub.Err()
+		defer func() {
+			sub.Unsubscribe()
+			for {
+				select {
+				case <-headEventCh: // drain channel
+				default:
+					return
+				}
+			}
+		}()
+	}
+
 	for {
 		select {
 		case ann := <-f.notify:
@@ -837,6 +878,24 @@ func (f *TxFetcher) loop() {
 				f.rescheduleTimeout(timeoutTimer, timeoutTrigger)
 			}
 
+		case ev := <-headEventCh:
+			// New head(s) added
+			newHead := ev.Header
+			if oldHead != nil && newHead.ParentHash != oldHead.Hash() {
+				// Reorg or setHead detected, clear the cache. We could be smarter here and
+				// only remove/add the diff, but this is simpler and not being exact here
+				// only results in a few more fetches.
+				f.txOnChainCache.Purge()
+			}
+			oldHead = newHead
+			// Add all transactions from the new block to the on-chain cache
+			for _, tx := range ev.Transactions {
+				f.txOnChainCache.Add(tx.Hash(), struct{}{})
+			}
+
+		case <-subErr:
+			return
+
 		case <-f.quit:
 			return
 		}

eth/fetcher/tx_fetcher_test.go

@@ -91,6 +91,7 @@ type txFetcherTest struct {
 // and deterministic randomness.
 func newTestTxFetcher() *TxFetcher {
 	return NewTxFetcher(
+		nil,
 		func(common.Hash, byte) error { return nil },
 		func(txs []*types.Transaction) []error {
 			return make([]error, len(txs))
@@ -2191,6 +2192,7 @@ func TestTransactionForgotten(t *testing.T) {
 	}
 
 	fetcher := NewTxFetcherForTests(
+		nil,
 		func(common.Hash, byte) error { return nil },
 		func(txs []*types.Transaction) []error {
 			errs := make([]error, len(txs))

eth/handler.go

@@ -29,7 +29,6 @@ import (
 
 	"github.com/dchest/siphash"
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/txpool"
@@ -59,11 +58,6 @@ const (
 	// All transactions with a higher size will be announced and need to be fetched
 	// by the peer.
 	txMaxBroadcastSize = 4096
-
-	// txOnChainCacheLimit is number of on-chain transactions to keep in a cache to avoid
-	// re-fetching them soon after they are mined.
-	// Approx 1MB for 30 minutes of transactions at 18 tps
-	txOnChainCacheLimit = 32768
 )
 
 var syncChallengeTimeout = 15 * time.Second // Time allowance for a node to reply to the sync progress challenge
@@ -132,9 +126,6 @@ type handler struct {
 	peers          *peerSet
 	txBroadcastKey [16]byte
 
-	// cache of on-chain transactions to avoid fetching once the tx gets on chain
-	txOnChainCache *lru.Cache[common.Hash, struct{}]
-
 	eventMux   *event.TypeMux
 	txsCh      chan core.NewTxsEvent
 	txsSub     event.Subscription
@@ -192,16 +183,12 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		if h.txpool.Has(tx) {
 			return txpool.ErrAlreadyKnown
 		}
-		// check on chain as well (no need to check limbo separately, as chain checks limbo too)
-		if _, exist := h.txOnChainCache.Get(tx); exist {
-			return core.ErrNonceTooLow
-		}
 		if !h.txpool.FilterType(kind) {
 			return types.ErrTxTypeNotSupported
 		}
 		return nil
 	}
-	h.txFetcher = fetcher.NewTxFetcher(validateMeta, addTxs, fetchTx, h.removePeer)
+	h.txFetcher = fetcher.NewTxFetcher(h.chain, validateMeta, addTxs, fetchTx, h.removePeer)
 	return h, nil
 }
 
@@ -436,10 +423,6 @@ func (h *handler) Start(maxPeers int) {
 	h.blockRange = newBlockRangeState(h.chain, h.eventMux)
 	go h.blockRangeLoop(h.blockRange)
 
-	h.wg.Add(1)
-	h.txOnChainCache = lru.NewCache[common.Hash, struct{}](txOnChainCacheLimit)
-	go h.txOnChainCacheLoop()
-
 	// start sync handlers
 	h.txFetcher.Start()
 
@@ -731,49 +714,3 @@ func (bc *broadcastChoice) choosePeers(peers []*ethPeer, txSender common.Address
 	}
 	return bc.buffer
 }
-
-// txOnChainCacheLoop listens to new chain events and updates the on-chain tx cache,
-// used as part of the filter on incoming tx announcements. In case of a reorg, the
-// cache is reset.
-func (h *handler) txOnChainCacheLoop() {
-	defer h.wg.Done()
-	//Subscribe to chain events to know when transactions are added to chain
-	headEventCh := make(chan core.ChainEvent, 10)
-	sub := h.chain.SubscribeChainEvent(headEventCh)
-	defer func() {
-		sub.Unsubscribe()
-		// drain channel
-		for {
-			select {
-			case <-headEventCh:
-			default:
-				return
-			}
-		}
-	}()
-
-	var oldHead *types.Header
-	for {
-		select {
-		case ev := <-headEventCh:
-			// New head(s) added
-			newHead := ev.Header
-			if oldHead != nil && newHead.ParentHash != oldHead.Hash() {
-				// Reorg or setHead detected, clear the cache. We could be smarter here and
-				// only remove/add the diff, but this is simpler and not being exact here
-				// only results in a few more fetches.
-				h.txOnChainCache.Purge()
-			}
-			oldHead = newHead
-			// Add all transactions from the new block to the on-chain cache
-			for _, tx := range ev.Transactions {
-				h.txOnChainCache.Add(tx.Hash(), struct{}{})
-			}
-		// exit when the subscription ends
-		case <-sub.Err():
-			return
-		case <-h.quitSync:
-			return
-		}
-	}
-}

tests/fuzzers/txfetcher/txfetcher_fuzzer.go

@@ -78,6 +78,7 @@ func fuzz(input []byte) int {
 	rand := rand.New(rand.NewSource(0x3a29)) // Same used in package tests!!!
 
 	f := fetcher.NewTxFetcherForTests(
+		nil,
 		func(common.Hash, byte) error { return nil },
 		func(txs []*types.Transaction) []error {
 			return make([]error, len(txs))