forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-21 20:26:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/fetcher: add docs to helper, use consistent naming

Browse source
This commit is contained in:
lightclient 2026-01-15 10:34:41 -07:00
parent 8f2e27b3be
commit 21171e07f3
No known key found for this signature in database
GPG key ID: 657913021EF45A6A
1 changed files with 30 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
eth/fetcher/tx_fetcher_test.go
View file

@ -87,7 +87,9 @@ type txFetcherTest struct {
steps []interface{} steps []interface{}
} }
func initDefaultTxFetcher() *TxFetcher { // newTestTxFetcher creates a tx fetcher with noop callbacks, simulated clock,
// and deterministic randomness.
func newTestTxFetcher() *TxFetcher {
return NewTxFetcher( return NewTxFetcher(
func(common.Hash, byte) error { return nil }, func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error { func(txs []*types.Transaction) []error {
@ -106,7 +108,7 @@ func initDefaultTxFetcher() *TxFetcher {
// with all the useless extra fields. // with all the useless extra fields.
func TestTransactionFetcherWaiting(t *testing.T) { func TestTransactionFetcherWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Initial announcement to get something into the waitlist // 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}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}},
@ -301,7 +303,7 @@ func TestTransactionFetcherWaiting(t *testing.T) {
// already scheduled. // already scheduled.
func TestTransactionFetcherSkipWaiting(t *testing.T) { func TestTransactionFetcherSkipWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // Push an initial announcement through to the scheduled stage
doTxNotify{ doTxNotify{
@ -384,7 +386,7 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) {
// and subsequent announces block or get allotted to someone else. // and subsequent announces block or get allotted to someone else.
func TestTransactionFetcherSingletonRequesting(t *testing.T) { func TestTransactionFetcherSingletonRequesting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // 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}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}},
@ -483,7 +485,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) {
proceed := make(chan struct{}) proceed := make(chan struct{})
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher { init: func() *TxFetcher {
f := initDefaultTxFetcher() f := newTestTxFetcher()
f.fetchTxs = func(origin string, hashes []common.Hash) error { f.fetchTxs = func(origin string, hashes []common.Hash) error {
<-proceed <-proceed
return errors.New("peer disconnected") return errors.New("peer disconnected")
@ -563,7 +565,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) {
// are cleaned up. // are cleaned up.
func TestTransactionFetcherCleanup(t *testing.T) { func TestTransactionFetcherCleanup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // 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())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
@ -598,7 +600,7 @@ func TestTransactionFetcherCleanup(t *testing.T) {
// this was a bug)). // this was a bug)).
func TestTransactionFetcherCleanupEmpty(t *testing.T) { func TestTransactionFetcherCleanupEmpty(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // 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())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
@ -632,7 +634,7 @@ func TestTransactionFetcherCleanupEmpty(t *testing.T) {
// different peer, or self if they are after the cutoff point. // different peer, or self if they are after the cutoff point.
func TestTransactionFetcherMissingRescheduling(t *testing.T) { func TestTransactionFetcherMissingRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // Push an initial announcement through to the scheduled stage
doTxNotify{peer: "A", doTxNotify{peer: "A",
@ -684,7 +686,7 @@ func TestTransactionFetcherMissingRescheduling(t *testing.T) {
// delivered, the peer gets properly cleaned out from the internal state. // delivered, the peer gets properly cleaned out from the internal state.
func TestTransactionFetcherMissingCleanup(t *testing.T) { func TestTransactionFetcherMissingCleanup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // Push an initial announcement through to the scheduled stage
doTxNotify{peer: "A", doTxNotify{peer: "A",
@ -724,7 +726,7 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) {
// Tests that transaction broadcasts properly clean up announcements. // Tests that transaction broadcasts properly clean up announcements.
func TestTransactionFetcherBroadcasts(t *testing.T) { func TestTransactionFetcherBroadcasts(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Set up three transactions to be in different stats, waiting, queued and fetching // 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())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
@ -771,7 +773,7 @@ func TestTransactionFetcherBroadcasts(t *testing.T) {
// Tests that the waiting list timers properly reset and reschedule. // Tests that the waiting list timers properly reset and reschedule.
func TestTransactionFetcherWaitTimerResets(t *testing.T) { func TestTransactionFetcherWaitTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
isWaiting(map[string][]announce{ isWaiting(map[string][]announce{
@ -834,7 +836,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) {
// out and be re-scheduled for someone else. // out and be re-scheduled for someone else.
func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { func TestTransactionFetcherTimeoutRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Push an initial announcement through to the scheduled stage // Push an initial announcement through to the scheduled stage
doTxNotify{ doTxNotify{
@ -903,7 +905,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) {
// Tests that the fetching timeout timers properly reset and reschedule. // Tests that the fetching timeout timers properly reset and reschedule.
func TestTransactionFetcherTimeoutTimerResets(t *testing.T) { func TestTransactionFetcherTimeoutTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
doWait{time: txArriveTimeout, step: true}, doWait{time: txArriveTimeout, step: true},
@ -974,7 +976,7 @@ func TestTransactionFetcherRateLimiting(t *testing.T) {
}) })
} }
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Announce all the transactions, wait a bit and ensure only a small // Announce all the transactions, wait a bit and ensure only a small
// percentage gets requested // percentage gets requested
@ -997,7 +999,7 @@ func TestTransactionFetcherRateLimiting(t *testing.T) {
// be requested at a time, to keep the responses below a reasonable level. // be requested at a time, to keep the responses below a reasonable level.
func TestTransactionFetcherBandwidthLimiting(t *testing.T) { func TestTransactionFetcherBandwidthLimiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Announce mid size transactions from A to verify that multiple // Announce mid size transactions from A to verify that multiple
// ones can be piled into a single request. // ones can be piled into a single request.
@ -1107,7 +1109,7 @@ func TestTransactionFetcherDoSProtection(t *testing.T) {
}) })
} }
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Announce half of the transaction and wait for them to be scheduled // 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]}, doTxNotify{peer: "A", hashes: hashesA[:maxTxAnnounces/2], types: typesA[:maxTxAnnounces/2], sizes: sizesA[:maxTxAnnounces/2]},
@ -1168,7 +1170,7 @@ func TestTransactionFetcherDoSProtection(t *testing.T) {
func TestTransactionFetcherUnderpricedDedup(t *testing.T) { func TestTransactionFetcherUnderpricedDedup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher { init: func() *TxFetcher {
f := initDefaultTxFetcher() f := newTestTxFetcher()
f.addTxs = func(txs []*types.Transaction) []error { f.addTxs = func(txs []*types.Transaction) []error {
errs := make([]error, len(txs)) errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ { for i := 0; i < len(errs); i++ {
@ -1266,7 +1268,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) {
} }
testTransactionFetcher(t, txFetcherTest{ testTransactionFetcher(t, txFetcherTest{
init: func() *TxFetcher { init: func() *TxFetcher {
f := initDefaultTxFetcher() f := newTestTxFetcher()
f.addTxs = func(txs []*types.Transaction) []error { f.addTxs = func(txs []*types.Transaction) []error {
errs := make([]error, len(txs)) errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ { for i := 0; i < len(errs); i++ {
@ -1294,7 +1296,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) {
// Tests that unexpected deliveries don't corrupt the internal state. // Tests that unexpected deliveries don't corrupt the internal state.
func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Deliver something out of the blue // Deliver something out of the blue
isWaiting(nil), isWaiting(nil),
@ -1344,7 +1346,7 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) {
// live or dangling stages. // live or dangling stages.
func TestTransactionFetcherDrop(t *testing.T) { func TestTransactionFetcherDrop(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Set up a few hashes into various stages // Set up a few hashes into various stages
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
@ -1409,7 +1411,7 @@ func TestTransactionFetcherDrop(t *testing.T) {
// available peer. // available peer.
func TestTransactionFetcherDropRescheduling(t *testing.T) { func TestTransactionFetcherDropRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Set up a few hashes into various stages // Set up a few hashes into various stages
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}},
@ -1447,7 +1449,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) {
drop := make(chan string, 2) drop := make(chan string, 2)
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher { init: func() *TxFetcher {
f := initDefaultTxFetcher() f := newTestTxFetcher()
f.dropPeer = func(peer string) { drop <- peer } f.dropPeer = func(peer string) { drop <- peer }
return f return f
}, },
@ -1524,7 +1526,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) {
// announced one. // announced one.
func TestTransactionFetcherFuzzCrash01(t *testing.T) { func TestTransactionFetcherFuzzCrash01(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Get a transaction into fetching mode and make it dangling with a broadcast // 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())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
@ -1543,7 +1545,7 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) {
// concurrently announced one. // concurrently announced one.
func TestTransactionFetcherFuzzCrash02(t *testing.T) { func TestTransactionFetcherFuzzCrash02(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Get a transaction into fetching mode and make it dangling with a broadcast // 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())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
@ -1564,7 +1566,7 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) {
// with a concurrent notify. // with a concurrent notify.
func TestTransactionFetcherFuzzCrash03(t *testing.T) { func TestTransactionFetcherFuzzCrash03(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Get a transaction into fetching mode and make it dangling with a broadcast // Get a transaction into fetching mode and make it dangling with a broadcast
doTxNotify{ doTxNotify{
@ -1595,7 +1597,7 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher { init: func() *TxFetcher {
f := initDefaultTxFetcher() f := newTestTxFetcher()
f.fetchTxs = func(string, []common.Hash) error { f.fetchTxs = func(string, []common.Hash) error {
<-proceed <-proceed
return errors.New("peer disconnected") return errors.New("peer disconnected")
@ -1624,7 +1626,7 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) {
// once they are announced in the network. // once they are announced in the network.
func TestBlobTransactionAnnounce(t *testing.T) { func TestBlobTransactionAnnounce(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
// Initial announcement to get something into the waitlist // 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}}, doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}},
@ -1685,7 +1687,7 @@ func TestBlobTransactionAnnounce(t *testing.T) {
func TestTransactionFetcherDropAlternates(t *testing.T) { func TestTransactionFetcherDropAlternates(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{ testTransactionFetcherParallel(t, txFetcherTest{
init: initDefaultTxFetcher, init: newTestTxFetcher,
steps: []interface{}{ steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
doWait{time: txArriveTimeout, step: true}, doWait{time: txArriveTimeout, step: true},