From b6014b3060e9567addee923252d8e6074b9fa72e Mon Sep 17 00:00:00 2001 From: Csaba Kiraly Date: Thu, 8 Jan 2026 09:52:22 +0100 Subject: [PATCH] simplify to use new cache only Signed-off-by: Csaba Kiraly --- core/blockchain_reader.go | 26 +-- core/rawdb/accessors_indexes.go | 5 - core/rawdb/accessors_indexes_test.go | 23 -- eth/fetcher/tx_fetcher.go | 1 + eth/fetcher/tx_fetcher_test.go | 330 +++++++++++++++++++++------ eth/handler.go | 2 +- 6 files changed, 266 insertions(+), 121 deletions(-) diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 5bafc02ba3..56a9818f5f 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -321,28 +321,12 @@ func (bc *BlockChain) GetAncestor(hash common.Hash, number, ancestor uint64, max return bc.hc.GetAncestor(hash, number, ancestor, maxNonCanonical) } -// HasCanonicalTransaction is a lightweight check to see if a transaction is present -// in the indexed part of the canonical chain without retrieving the transaction itself. -// It's view is limited to the indexed part of the chain, so very old transactions -// might fail the check if the indexer was constrained, or indexing is still in progress. -// The cacheOnly flag restricts the check to the in-memory cache, avoiding database -// access altogether. -func (bc *BlockChain) HasCanonicalTransaction(hash common.Hash, cacheOnly bool) bool { +// TxInCanonicalCache is a lightweight check to see if a transaction is present in +// the cached part of the canonical chain without retrieving the transaction itself. +func (bc *BlockChain) TxInCanonicalCache(hash common.Hash) bool { // Check in memory cache first - if _, exist := bc.txOnChainCache.Get(hash); exist { - return true - } - - // Check in memory tx cache next, without bumping the LRU - if bc.txLookupCache.Contains(hash) { - return true - } - - // Fallback to database lookup, without reading the transaction itself - if !cacheOnly && rawdb.HasCanonicalTransaction(bc.db, hash) { - return true - } - return false + _, exist := bc.txOnChainCache.Get(hash) + return exist } // GetCanonicalTransaction retrieves the lookup along with the transaction diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 107448a2e2..10eb454015 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -174,11 +174,6 @@ func findTxInBlockBody(blockbody rlp.RawValue, target common.Hash) (*types.Trans return nil, 0, errors.New("transaction not found") } -// HasCanonicalTransaction checks whether a specific transaction is in the database. -func HasCanonicalTransaction(db ethdb.Reader, hash common.Hash) bool { - return ReadTxLookupEntry(db, hash) != nil -} - // ReadCanonicalTransaction retrieves a specific transaction from the database, along // with its added positional metadata. Notably, only the transaction in the canonical // chain is visible. diff --git a/core/rawdb/accessors_indexes_test.go b/core/rawdb/accessors_indexes_test.go index c47516ea84..a812fefeaa 100644 --- a/core/rawdb/accessors_indexes_test.go +++ b/core/rawdb/accessors_indexes_test.go @@ -89,10 +89,6 @@ func TestLookupStorage(t *testing.T) { // Check that no transactions entries are in a pristine database for i, tx := range txs { - // Sometimes (but not always) check HasCanonicalTransaction as well - if i%2 == 0 && HasCanonicalTransaction(db, tx.Hash()) { - t.Fatalf("tx #%d [%x]: non existent transaction found", i, tx.Hash()) - } if txn, _, _, _ := ReadCanonicalTransaction(db, tx.Hash()); txn != nil { t.Fatalf("tx #%d [%x]: non existent transaction returned: %v", i, tx.Hash(), txn) } @@ -103,9 +99,6 @@ func TestLookupStorage(t *testing.T) { tc.writeTxLookupEntriesByBlock(db, block) for i, tx := range txs { - if i%2 == 0 && !HasCanonicalTransaction(db, tx.Hash()) { - t.Fatalf("tx #%d [%x]: transaction not found", i, tx.Hash()) - } if txn, hash, number, index := ReadCanonicalTransaction(db, tx.Hash()); txn == nil { t.Fatalf("tx #%d [%x]: transaction not found", i, tx.Hash()) } else { @@ -120,26 +113,10 @@ func TestLookupStorage(t *testing.T) { // Delete the transactions and check purge for i, tx := range txs { DeleteTxLookupEntry(db, tx.Hash()) - if i%2 == 0 && HasCanonicalTransaction(db, tx.Hash()) { - t.Fatalf("tx #%d [%x]: deleted transaction found", i, tx.Hash()) - } if txn, _, _, _ := ReadCanonicalTransaction(db, tx.Hash()); txn != nil { t.Fatalf("tx #%d [%x]: deleted transaction returned: %v", i, tx.Hash(), txn) } } - // Check some random hashes as well - for i := 0; i < 10; i++ { - var hash common.Hash - for j := 0; j < len(hash); j++ { - hash[j] = byte(i*13 + j*7) - } - if i%2 == 0 && HasCanonicalTransaction(db, hash) { - t.Fatalf("random tx %d [%x]: non existent transaction found", i, hash) - } - if txn, _, _, _ := ReadCanonicalTransaction(db, hash); txn != nil { - t.Fatalf("random tx %d [%x]: non existent transaction returned: %v", i, hash, txn) - } - } }) } } diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 6a93eea1e5..f024f3aeba 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -320,6 +320,7 @@ func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) 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 diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index a6253ada26..d6d5a8692e 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/eth/fetcher/tx_fetcher_test.go @@ -83,17 +83,6 @@ type txFetcherTest struct { steps []interface{} } -func initDefaultTxFetcher() *TxFetcher { - return NewTxFetcher( - func(common.Hash, byte) error { return nil }, - func(txs []*types.Transaction) []error { - return make([]error, len(txs)) - }, - func(string, []common.Hash) error { return nil }, - nil, - ) -} - // Tests that transaction announcements with associated metadata are added to a // waitlist, and none of them are scheduled for retrieval until the wait expires. // @@ -102,7 +91,14 @@ func initDefaultTxFetcher() *TxFetcher { // with all the useless extra fields. func TestTransactionFetcherWaiting(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Initial announcement to get something into the waitlist doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}}, @@ -297,7 +293,14 @@ func TestTransactionFetcherWaiting(t *testing.T) { // already scheduled. func TestTransactionFetcherSkipWaiting(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{ @@ -380,7 +383,14 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) { // and subsequent announces block or get allotted to someone else. func TestTransactionFetcherSingletonRequesting(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}}, @@ -479,12 +489,15 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) { proceed := make(chan struct{}) testTransactionFetcherParallel(t, txFetcherTest{ init: func() *TxFetcher { - f := initDefaultTxFetcher() - f.fetchTxs = func(origin string, hashes []common.Hash) error { - <-proceed - return errors.New("peer disconnected") - } - return f + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(origin string, hashes []common.Hash) error { + <-proceed + return errors.New("peer disconnected") + }, + nil, + ) }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage @@ -559,7 +572,16 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) { // are cleaned up. func TestTransactionFetcherCleanup(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, @@ -594,7 +616,16 @@ func TestTransactionFetcherCleanup(t *testing.T) { // this was a bug)). func TestTransactionFetcherCleanupEmpty(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, @@ -628,7 +659,16 @@ func TestTransactionFetcherCleanupEmpty(t *testing.T) { // different peer, or self if they are after the cutoff point. func TestTransactionFetcherMissingRescheduling(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{peer: "A", @@ -680,7 +720,16 @@ func TestTransactionFetcherMissingRescheduling(t *testing.T) { // delivered, the peer gets properly cleaned out from the internal state. func TestTransactionFetcherMissingCleanup(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{peer: "A", @@ -720,7 +769,16 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) { // Tests that transaction broadcasts properly clean up announcements. func TestTransactionFetcherBroadcasts(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Set up three transactions to be in different stats, waiting, queued and fetching doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, @@ -767,7 +825,14 @@ func TestTransactionFetcherBroadcasts(t *testing.T) { // Tests that the waiting list timers properly reset and reschedule. func TestTransactionFetcherWaitTimerResets(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, isWaiting(map[string][]announce{ @@ -830,7 +895,16 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) { // out and be re-scheduled for someone else. func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Push an initial announcement through to the scheduled stage doTxNotify{ @@ -899,7 +973,14 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { // Tests that the fetching timeout timers properly reset and reschedule. func TestTransactionFetcherTimeoutTimerResets(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, doWait{time: txArriveTimeout, step: true}, @@ -970,7 +1051,14 @@ func TestTransactionFetcherRateLimiting(t *testing.T) { }) } testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Announce all the transactions, wait a bit and ensure only a small // percentage gets requested @@ -993,7 +1081,14 @@ func TestTransactionFetcherRateLimiting(t *testing.T) { // be requested at a time, to keep the responses below a reasonable level. func TestTransactionFetcherBandwidthLimiting(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Announce mid size transactions from A to verify that multiple // ones can be piled into a single request. @@ -1103,7 +1198,14 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { }) } testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Announce half of the transaction and wait for them to be scheduled doTxNotify{peer: "A", hashes: hashesA[:maxTxAnnounces/2], types: typesA[:maxTxAnnounces/2], sizes: sizesA[:maxTxAnnounces/2]}, @@ -1164,21 +1266,24 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { func TestTransactionFetcherUnderpricedDedup(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ init: func() *TxFetcher { - f := initDefaultTxFetcher() - f.addTxs = func(txs []*types.Transaction) []error { - errs := make([]error, len(txs)) - for i := 0; i < len(errs); i++ { - if i%3 == 0 { - errs[i] = txpool.ErrUnderpriced - } else if i%3 == 1 { - errs[i] = txpool.ErrReplaceUnderpriced - } else { - errs[i] = txpool.ErrTxGasPriceTooLow + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + errs := make([]error, len(txs)) + for i := 0; i < len(errs); i++ { + if i%3 == 0 { + errs[i] = txpool.ErrUnderpriced + } else if i%3 == 1 { + errs[i] = txpool.ErrReplaceUnderpriced + } else { + errs[i] = txpool.ErrTxGasPriceTooLow + } } - } - return errs - } - return f + return errs + }, + func(string, []common.Hash) error { return nil }, + nil, + ) }, steps: []interface{}{ // Deliver a transaction through the fetcher, but reject as underpriced @@ -1262,15 +1367,18 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) { } testTransactionFetcher(t, txFetcherTest{ init: func() *TxFetcher { - f := initDefaultTxFetcher() - f.addTxs = func(txs []*types.Transaction) []error { - errs := make([]error, len(txs)) - for i := 0; i < len(errs); i++ { - errs[i] = txpool.ErrUnderpriced - } - return errs - } - return f + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + errs := make([]error, len(txs)) + for i := 0; i < len(errs); i++ { + errs[i] = txpool.ErrUnderpriced + } + return errs + }, + func(string, []common.Hash) error { return nil }, + nil, + ) }, steps: append(steps, []interface{}{ // The preparation of the test has already been done in `steps`, add the last check @@ -1290,7 +1398,16 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) { // Tests that unexpected deliveries don't corrupt the internal state. func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Deliver something out of the blue isWaiting(nil), @@ -1340,7 +1457,16 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { // live or dangling stages. func TestTransactionFetcherDrop(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Set up a few hashes into various stages doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, @@ -1405,7 +1531,16 @@ func TestTransactionFetcherDrop(t *testing.T) { // available peer. func TestTransactionFetcherDropRescheduling(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Set up a few hashes into various stages doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, @@ -1443,9 +1578,14 @@ func TestInvalidAnnounceMetadata(t *testing.T) { drop := make(chan string, 2) testTransactionFetcherParallel(t, txFetcherTest{ init: func() *TxFetcher { - f := initDefaultTxFetcher() - f.dropPeer = func(peer string) { drop <- peer } - return f + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + func(peer string) { drop <- peer }, + ) }, steps: []interface{}{ // Initial announcement to get something into the waitlist @@ -1520,7 +1660,16 @@ func TestInvalidAnnounceMetadata(t *testing.T) { // announced one. func TestTransactionFetcherFuzzCrash01(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Get a transaction into fetching mode and make it dangling with a broadcast doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, @@ -1539,7 +1688,16 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) { // concurrently announced one. func TestTransactionFetcherFuzzCrash02(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Get a transaction into fetching mode and make it dangling with a broadcast doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, @@ -1560,7 +1718,16 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) { // with a concurrent notify. func TestTransactionFetcherFuzzCrash03(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Get a transaction into fetching mode and make it dangling with a broadcast doTxNotify{ @@ -1591,12 +1758,17 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ init: func() *TxFetcher { - f := initDefaultTxFetcher() - f.fetchTxs = func(string, []common.Hash) error { - <-proceed - return errors.New("peer disconnected") - } - return f + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { + <-proceed + return errors.New("peer disconnected") + }, + nil, + ) }, steps: []interface{}{ // Get a transaction into fetching mode and make it dangling with a broadcast @@ -1620,7 +1792,14 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) { // once they are announced in the network. func TestBlobTransactionAnnounce(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ // Initial announcement to get something into the waitlist doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}}, @@ -1681,7 +1860,16 @@ func TestBlobTransactionAnnounce(t *testing.T) { func TestTransactionFetcherDropAlternates(t *testing.T) { testTransactionFetcherParallel(t, txFetcherTest{ - init: initDefaultTxFetcher, + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash, byte) error { return nil }, + func(txs []*types.Transaction) []error { + return make([]error, len(txs)) + }, + func(string, []common.Hash) error { return nil }, + nil, + ) + }, steps: []interface{}{ doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, doWait{time: txArriveTimeout, step: true}, diff --git a/eth/handler.go b/eth/handler.go index 8a7b3a581c..a077df4040 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -184,7 +184,7 @@ func newHandler(config *handlerConfig) (*handler, error) { return txpool.ErrAlreadyKnown } // check on chain as well (no need to check limbo separately, as chain checks limbo too) - if h.chain.HasCanonicalTransaction(hash, true) { + if h.chain.TxInCanonicalCache(hash) { return core.ErrNonceTooLow } if !h.txpool.FilterType(kind) {