simplify to use new cache only

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

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
+// TxInCanonicalCache is a lightweight check to see if a transaction is present in
-// in the indexed part of the canonical chain without retrieving the transaction itself.
+// the cached 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
+func (bc *BlockChain) TxInCanonicalCache(hash common.Hash) bool {
-// 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 {
 	// Check in memory cache first
-	if _, exist := bc.txOnChainCache.Get(hash); exist {
+	_, exist := bc.txOnChainCache.Get(hash)
-		return true
+	return exist
-	}
-
-	// 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
 }
 
 // GetCanonicalTransaction retrieves the lookup along with the transaction

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.

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)
-				}
-			}
 		})
 	}
 }

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

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()
+			return NewTxFetcher(
-			f.fetchTxs = func(origin string, hashes []common.Hash) error {
+				func(common.Hash, byte) error { return nil },
-				<-proceed
+				nil,
-				return errors.New("peer disconnected")
+				func(origin string, hashes []common.Hash) error {
-			}
+					<-proceed
-			return f
+					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()
+			return NewTxFetcher(
-			f.addTxs = func(txs []*types.Transaction) []error {
+				func(common.Hash, byte) error { return nil },
-				errs := make([]error, len(txs))
+				func(txs []*types.Transaction) []error {
-				for i := 0; i < len(errs); i++ {
+					errs := make([]error, len(txs))
-					if i%3 == 0 {
+					for i := 0; i < len(errs); i++ {
-						errs[i] = txpool.ErrUnderpriced
+						if i%3 == 0 {
-					} else if i%3 == 1 {
+							errs[i] = txpool.ErrUnderpriced
-						errs[i] = txpool.ErrReplaceUnderpriced
+						} else if i%3 == 1 {
-					} else {
+							errs[i] = txpool.ErrReplaceUnderpriced
-						errs[i] = txpool.ErrTxGasPriceTooLow
+						} else {
+							errs[i] = txpool.ErrTxGasPriceTooLow
+						}
 					}
-				}
+					return errs
-				return errs
+				},
-			}
+				func(string, []common.Hash) error { return nil },
-			return f
+				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()
+			return NewTxFetcher(
-			f.addTxs = func(txs []*types.Transaction) []error {
+				func(common.Hash, byte) error { return nil },
-				errs := make([]error, len(txs))
+				func(txs []*types.Transaction) []error {
-				for i := 0; i < len(errs); i++ {
+					errs := make([]error, len(txs))
-					errs[i] = txpool.ErrUnderpriced
+					for i := 0; i < len(errs); i++ {
-				}
+						errs[i] = txpool.ErrUnderpriced
-				return errs
+					}
-			}
+					return errs
-			return f
+				},
+				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()
+			return NewTxFetcher(
-			f.dropPeer = func(peer string) { drop <- peer }
+				func(common.Hash, byte) error { return nil },
-			return f
+				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()
+			return NewTxFetcher(
-			f.fetchTxs = func(string, []common.Hash) error {
+				func(common.Hash, byte) error { return nil },
-				<-proceed
+				func(txs []*types.Transaction) []error {
-				return errors.New("peer disconnected")
+					return make([]error, len(txs))
-			}
+				},
-			return f
+				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},

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) {