eth/fetcher: refactor test code

Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>
This commit is contained in:
Csaba Kiraly 2026-01-15 17:15:51 +01:00
parent 494908a852
commit 8f2e27b3be
No known key found for this signature in database
GPG key ID: 0FE274EE8C95166E

View file

@ -87,6 +87,17 @@ 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.
//
@ -95,14 +106,7 @@ type txFetcherTest struct {
// with all the useless extra fields.
func TestTransactionFetcherWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
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,14 +301,7 @@ func TestTransactionFetcherWaiting(t *testing.T) {
// already scheduled.
func TestTransactionFetcherSkipWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Push an initial announcement through to the scheduled stage
doTxNotify{
@ -387,14 +384,7 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) {
// and subsequent announces block or get allotted to someone else.
func TestTransactionFetcherSingletonRequesting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
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}},
@ -493,15 +483,12 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) {
proceed := make(chan struct{})
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(origin string, hashes []common.Hash) error {
<-proceed
return errors.New("peer disconnected")
},
nil,
)
f := initDefaultTxFetcher()
f.fetchTxs = func(origin string, hashes []common.Hash) error {
<-proceed
return errors.New("peer disconnected")
}
return f
},
steps: []interface{}{
// Push an initial announcement through to the scheduled stage
@ -576,16 +563,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) {
// are cleaned up.
func TestTransactionFetcherCleanup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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())}},
@ -620,16 +598,7 @@ func TestTransactionFetcherCleanup(t *testing.T) {
// this was a bug)).
func TestTransactionFetcherCleanupEmpty(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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())}},
@ -663,16 +632,7 @@ 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: 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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Push an initial announcement through to the scheduled stage
doTxNotify{peer: "A",
@ -724,16 +684,7 @@ 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: 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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Push an initial announcement through to the scheduled stage
doTxNotify{peer: "A",
@ -773,16 +724,7 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) {
// Tests that transaction broadcasts properly clean up announcements.
func TestTransactionFetcherBroadcasts(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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())}},
@ -829,14 +771,7 @@ func TestTransactionFetcherBroadcasts(t *testing.T) {
// Tests that the waiting list timers properly reset and reschedule.
func TestTransactionFetcherWaitTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
isWaiting(map[string][]announce{
@ -899,16 +834,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) {
// out and be re-scheduled for someone else.
func TestTransactionFetcherTimeoutRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Push an initial announcement through to the scheduled stage
doTxNotify{
@ -977,14 +903,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) {
// Tests that the fetching timeout timers properly reset and reschedule.
func TestTransactionFetcherTimeoutTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
doWait{time: txArriveTimeout, step: true},
@ -1055,14 +974,7 @@ func TestTransactionFetcherRateLimiting(t *testing.T) {
})
}
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Announce all the transactions, wait a bit and ensure only a small
// percentage gets requested
@ -1085,14 +997,7 @@ 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: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Announce mid size transactions from A to verify that multiple
// ones can be piled into a single request.
@ -1202,14 +1107,7 @@ func TestTransactionFetcherDoSProtection(t *testing.T) {
})
}
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
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]},
@ -1270,24 +1168,21 @@ func TestTransactionFetcherDoSProtection(t *testing.T) {
func TestTransactionFetcherUnderpricedDedup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
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
}
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 errs
},
func(string, []common.Hash) error { return nil },
nil,
)
}
return errs
}
return f
},
steps: []interface{}{
// Deliver a transaction through the fetcher, but reject as underpriced
@ -1371,18 +1266,15 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) {
}
testTransactionFetcher(t, txFetcherTest{
init: func() *TxFetcher {
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,
)
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
},
steps: append(steps, []interface{}{
// The preparation of the test has already been done in `steps`, add the last check
@ -1402,16 +1294,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) {
// Tests that unexpected deliveries don't corrupt the internal state.
func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Deliver something out of the blue
isWaiting(nil),
@ -1461,16 +1344,7 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) {
// live or dangling stages.
func TestTransactionFetcherDrop(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Set up a few hashes into various stages
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
@ -1535,16 +1409,7 @@ func TestTransactionFetcherDrop(t *testing.T) {
// available peer.
func TestTransactionFetcherDropRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Set up a few hashes into various stages
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
@ -1582,14 +1447,9 @@ func TestInvalidAnnounceMetadata(t *testing.T) {
drop := make(chan string, 2)
testTransactionFetcherParallel(t, txFetcherTest{
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 },
func(peer string) { drop <- peer },
)
f := initDefaultTxFetcher()
f.dropPeer = func(peer string) { drop <- peer }
return f
},
steps: []interface{}{
// Initial announcement to get something into the waitlist
@ -1664,16 +1524,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) {
// announced one.
func TestTransactionFetcherFuzzCrash01(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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())}},
@ -1692,16 +1543,7 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) {
// concurrently announced one.
func TestTransactionFetcherFuzzCrash02(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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())}},
@ -1722,16 +1564,7 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) {
// with a concurrent notify.
func TestTransactionFetcherFuzzCrash03(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
steps: []interface{}{
// Get a transaction into fetching mode and make it dangling with a broadcast
doTxNotify{
@ -1762,17 +1595,12 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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 {
<-proceed
return errors.New("peer disconnected")
},
nil,
)
f := initDefaultTxFetcher()
f.fetchTxs = func(string, []common.Hash) error {
<-proceed
return errors.New("peer disconnected")
}
return f
},
steps: []interface{}{
// Get a transaction into fetching mode and make it dangling with a broadcast
@ -1796,14 +1624,7 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) {
// once they are announced in the network.
func TestBlobTransactionAnnounce(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
)
},
init: initDefaultTxFetcher,
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}},
@ -1864,16 +1685,7 @@ func TestBlobTransactionAnnounce(t *testing.T) {
func TestTransactionFetcherDropAlternates(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
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,
)
},
init: initDefaultTxFetcher,
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},