From 0ce494b60cd00d70f1f9f2dd0b9bfbd76204168a Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Fri, 19 Aug 2022 15:59:36 +0200 Subject: [PATCH] eth/fetcher: don't spend too much time on transaction inclusion (#25524) * eth/fetcher: introduce some lag in tx fetching * eth/fetcher: change conditions a bit * eth/fetcher: use per-batch quota check * eth/fetcher: fix some comments * eth/fetcher: address review concerns * eth/fetcher: fix panic + add warn log * eth/fetcher: fix log * eth/fetcher: fix log * cmd/devp2p/internal/ethtest: fix ignorign tx announcements from prev. tests * cmd/devp2p/internal/ethtest: fix TestLargeTxRequest This increases the number of tx relay messages the test waits for. Since go-ethereum now processes incoming txs in smaller batches, the announcement messages it sends are also smaller. Co-authored-by: Felix Lange --- cmd/devp2p/internal/ethtest/helpers.go | 6 +- cmd/devp2p/internal/ethtest/suite.go | 4 + cmd/devp2p/internal/ethtest/transaction.go | 6 +- eth/fetcher/tx_fetcher.go | 98 +++++++++++++--------- 4 files changed, 72 insertions(+), 42 deletions(-) diff --git a/cmd/devp2p/internal/ethtest/helpers.go b/cmd/devp2p/internal/ethtest/helpers.go index eeeb4f93cabf..b57649ade99d 100644 --- a/cmd/devp2p/internal/ethtest/helpers.go +++ b/cmd/devp2p/internal/ethtest/helpers.go @@ -357,9 +357,13 @@ func (s *Suite) waitAnnounce(conn *Conn, blockAnnouncement *NewBlock) error { return fmt.Errorf("wrong block hash in announcement: expected %v, got %v", blockAnnouncement.Block.Hash(), hashes[0].Hash) } return nil + + // ignore tx announcements from previous tests case *NewPooledTransactionHashes: - // ignore tx announcements from previous tests continue + case *Transactions: + continue + default: return fmt.Errorf("unexpected: %s", pretty.Sdump(msg)) } diff --git a/cmd/devp2p/internal/ethtest/suite.go b/cmd/devp2p/internal/ethtest/suite.go index 7059b4ba738c..4497478d72d6 100644 --- a/cmd/devp2p/internal/ethtest/suite.go +++ b/cmd/devp2p/internal/ethtest/suite.go @@ -544,9 +544,13 @@ func (s *Suite) TestNewPooledTxs(t *utesting.T) { t.Fatalf("unexpected number of txs requested: wanted %d, got %d", len(hashes), len(msg.GetPooledTransactionsPacket)) } return + // ignore propagated txs from previous tests case *NewPooledTransactionHashes: continue + case *Transactions: + continue + // ignore block announcements from previous tests case *NewBlockHashes: continue diff --git a/cmd/devp2p/internal/ethtest/transaction.go b/cmd/devp2p/internal/ethtest/transaction.go index c4748bf8f7d8..baa55bd49268 100644 --- a/cmd/devp2p/internal/ethtest/transaction.go +++ b/cmd/devp2p/internal/ethtest/transaction.go @@ -29,7 +29,7 @@ import ( "github.com/ethereum/go-ethereum/params" ) -//var faucetAddr = common.HexToAddress("0x71562b71999873DB5b286dF957af199Ec94617F7") +// var faucetAddr = common.HexToAddress("0x71562b71999873DB5b286dF957af199Ec94617F7") var faucetKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") func (s *Suite) sendSuccessfulTxs(t *utesting.T) error { @@ -192,10 +192,10 @@ func sendMultipleSuccessfulTxs(t *utesting.T, s *Suite, txs []*types.Transaction nonce = txs[len(txs)-1].Nonce() // Wait for the transaction announcement(s) and make sure all sent txs are being propagated. - // all txs should be announced within 3 announcements. + // all txs should be announced within a couple announcements. recvHashes := make([]common.Hash, 0) - for i := 0; i < 3; i++ { + for i := 0; i < 20; i++ { switch msg := recvConn.readAndServe(s.chain, timeout).(type) { case *Transactions: for _, tx := range *msg { diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 035e0c2ec7d8..7c8f16df531f 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -262,57 +262,79 @@ func (f *TxFetcher) Notify(peer string, hashes []common.Hash) error { // direct request replies. The differentiation is important so the fetcher can // re-schedule missing transactions as soon as possible. func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) error { - // Keep track of all the propagated transactions - if direct { - txReplyInMeter.Mark(int64(len(txs))) - } else { - txBroadcastInMeter.Mark(int64(len(txs))) + var ( + inMeter = txReplyInMeter + knownMeter = txReplyKnownMeter + underpricedMeter = txReplyUnderpricedMeter + otherRejectMeter = txReplyOtherRejectMeter + ) + if !direct { + inMeter = txBroadcastInMeter + knownMeter = txBroadcastKnownMeter + underpricedMeter = txBroadcastUnderpricedMeter + otherRejectMeter = txBroadcastOtherRejectMeter } + // Keep track of all the propagated transactions + inMeter.Mark(int64(len(txs))) + // Push all the transactions into the pool, tracking underpriced ones to avoid // re-requesting them and dropping the peer in case of malicious transfers. var ( - added = make([]common.Hash, 0, len(txs)) - duplicate int64 - underpriced int64 - otherreject int64 + added = make([]common.Hash, 0, len(txs)) + delay time.Duration ) - errs := f.addTxs(txs) - for i, err := range errs { - // Track the transaction hash if the price is too low for us. - // Avoid re-request this transaction when we receive another - // announcement. - if errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced) { - for f.underpriced.Cardinality() >= maxTxUnderpricedSetSize { - f.underpriced.Pop() - } - f.underpriced.Add(txs[i].Hash()) + // proceed in batches + for i := 0; i < len(txs); i += 128 { + end := i + 128 + if end > len(txs) { + end = len(txs) } - // Track a few interesting failure types - switch { - case err == nil: // Noop, but need to handle to not count these + var ( + duplicate int64 + underpriced int64 + otherreject int64 + ) + batch := txs[i:end] + for j, err := range f.addTxs(batch) { + // Track the transaction hash if the price is too low for us. + // Avoid re-request this transaction when we receive another + // announcement. + if errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced) { + for f.underpriced.Cardinality() >= maxTxUnderpricedSetSize { + f.underpriced.Pop() + } + f.underpriced.Add(batch[j].Hash()) + } + // Track a few interesting failure types + switch { + case err == nil: // Noop, but need to handle to not count these - case errors.Is(err, core.ErrAlreadyKnown): - duplicate++ + case errors.Is(err, core.ErrAlreadyKnown): + duplicate++ - case errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced): - underpriced++ + case errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced): + underpriced++ - default: - otherreject++ + default: + otherreject++ + } + added = append(added, batch[j].Hash()) + } + knownMeter.Mark(duplicate) + underpricedMeter.Mark(underpriced) + otherRejectMeter.Mark(otherreject) + + // If 'other reject' is >25% of the deliveries in any batch, abort. Either we are + // out of sync with the chain or the peer is griefing us. + if otherreject > 128/4 { + delay = 200 * time.Millisecond + log.Warn("Peer delivering useless transactions", "peer", peer, "ignored", len(txs)-end) + break } - added = append(added, txs[i].Hash()) - } - if direct { - txReplyKnownMeter.Mark(duplicate) - txReplyUnderpricedMeter.Mark(underpriced) - txReplyOtherRejectMeter.Mark(otherreject) - } else { - txBroadcastKnownMeter.Mark(duplicate) - txBroadcastUnderpricedMeter.Mark(underpriced) - txBroadcastOtherRejectMeter.Mark(otherreject) } select { case f.cleanup <- &txDelivery{origin: peer, hashes: added, direct: direct}: + time.Sleep(delay) return nil case <-f.quit: return errTerminated