diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 50d6f2f7ad..3f30a73254 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/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 { - return NewTxFetcherForTests(validateMeta, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil) +// Chain can be nil to disable on-chain checks. +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), - cleanup: make(chan *txDelivery), - drop: make(chan *txDrop), - quit: make(chan struct{}), - waitlist: make(map[common.Hash]map[string]struct{}), - waittime: make(map[common.Hash]mclock.AbsTime), - waitslots: make(map[string]map[common.Hash]*txMetadataWithSeq), - announces: make(map[string]map[common.Hash]*txMetadataWithSeq), - announced: make(map[common.Hash]map[string]struct{}), - fetching: make(map[common.Hash]string), - requests: make(map[string]*txRequest), - alternates: make(map[common.Hash]map[string]struct{}), - underpriced: lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize), - validateMeta: validateMeta, - addTxs: addTxs, - fetchTxs: fetchTxs, - dropPeer: dropPeer, - clock: clock, - realTime: realTime, - rand: rand, + notify: make(chan *txAnnounce), + cleanup: make(chan *txDelivery), + drop: make(chan *txDrop), + quit: make(chan struct{}), + waitlist: make(map[common.Hash]map[string]struct{}), + waittime: make(map[common.Hash]mclock.AbsTime), + waitslots: make(map[string]map[common.Hash]*txMetadataWithSeq), + announces: make(map[string]map[common.Hash]*txMetadataWithSeq), + announced: make(map[common.Hash]map[string]struct{}), + fetching: make(map[common.Hash]string), + requests: make(map[string]*txRequest), + alternates: make(map[common.Hash]map[string]struct{}), + underpriced: lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize), + txOnChainCache: lru.NewCache[common.Hash, struct{}](txOnChainCacheLimit), + chain: chain, + validateMeta: validateMeta, + addTxs: addTxs, + fetchTxs: fetchTxs, + dropPeer: dropPeer, + 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 } diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 87fbe9f38c..6c2719631e 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/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)) diff --git a/eth/handler.go b/eth/handler.go index fc1c55776a..bb2cd5f88b 100644 --- a/eth/handler.go +++ b/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 - } - } -} diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index 3baff33dcc..bcceaff383 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/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))