From 479a57d3d93822507470206b6bb3b18758b140cc Mon Sep 17 00:00:00 2001 From: Csaba Kiraly Date: Wed, 14 Jan 2026 19:02:03 +0100 Subject: [PATCH] eth: check announced transactions against chain as well The fetcher should not fetch transactions that are already on chain. Until now we were only checking in the txpool, but that does not have the old transaction. This was leading to extra fetches of transactions that were announced by a peer but are already on chain. Here we extend the check to the chain as well. Signed-off-by: Csaba Kiraly --- eth/handler.go | 53 ++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 46634cae88..26c084c0ed 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -29,6 +29,7 @@ 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" @@ -58,6 +59,11 @@ 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 @@ -126,6 +132,9 @@ 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 @@ -179,17 +188,19 @@ func newHandler(config *handlerConfig) (*handler, error) { addTxs := func(txs []*types.Transaction) []error { return h.txpool.Add(txs, false) } - validateMeta := func(tx common.Hash, kind byte) 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) return h, nil } @@ -425,6 +436,10 @@ 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() @@ -716,3 +731,37 @@ 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 sub.Unsubscribe() + + 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 + } + } +}