From 9332e19f6847791b470cda9afcdcb03300477fa3 Mon Sep 17 00:00:00 2001 From: Dmitry Kisel Date: Tue, 14 Jul 2026 10:48:27 +0100 Subject: [PATCH] eth, eth/fetcher, p2p: make tx arrival wait configurable via p2p.txarrivalwait The p2p.txarrivalwait config option and --txarrivalwait flag existed but were never wired into the transaction fetcher, which used the hardcoded 500ms txArriveTimeout since the v1.14.13 upstream merge. Plumb the value from p2p.Config through the eth handler into NewTxFetcher. A value of zero requests announced transactions immediately instead of waiting for a potential broadcast; negative values are clamped to zero. --- docs/cli/server.md | 2 +- eth/backend.go | 1 + eth/fetcher/tx_fetcher.go | 37 ++-- eth/fetcher/tx_fetcher_test.go | 187 ++++++++++++++++---- eth/handler.go | 3 +- eth/handler_test.go | 14 +- internal/cli/server/config.go | 3 +- internal/cli/server/config_test.go | 19 ++ internal/cli/server/flags.go | 2 +- p2p/config.go | 6 +- tests/fuzzers/txfetcher/txfetcher_fuzzer.go | 1 + 11 files changed, 220 insertions(+), 55 deletions(-) diff --git a/docs/cli/server.md b/docs/cli/server.md index 8c0fd5e44b..7949ef6cbe 100644 --- a/docs/cli/server.md +++ b/docs/cli/server.md @@ -322,7 +322,7 @@ The ```bor server``` command runs the Bor client. - ```txannouncementonly```: Whether to only announce transactions to peers (default: false) -- ```txarrivalwait```: Maximum duration to wait for a transaction before explicitly requesting it (default: 500ms) +- ```txarrivalwait```: Maximum duration to wait for a transaction before explicitly requesting it (0 requests announced transactions immediately) (default: 500ms) - ```v4disc```: Enables the V4 discovery mechanism (default: true) diff --git a/eth/backend.go b/eth/backend.go index d6414dd0b1..731d7c2424 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -455,6 +455,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { checker: checker, enableBlockTracking: eth.config.EnableBlockTracking, txAnnouncementOnly: eth.p2pServer.TxAnnouncementOnly, + txArrivalWait: eth.p2pServer.TxArrivalWait, disableTxPropagation: eth.p2pServer.DisableTxPropagation, witnessProtocol: eth.config.WitnessProtocol, syncWithWitnesses: eth.config.SyncWithWitnesses, diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 666d919549..34d1894d40 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -64,9 +64,9 @@ const ( // maxTxUnderpricedTimeout is the max time a transaction should be stuck in the underpriced set. maxTxUnderpricedTimeout = 5 * time.Minute - // txArriveTimeout is the time allowance before an announced transaction is - // explicitly requested. - txArriveTimeout = 500 * time.Millisecond + // DefaultTxArrivalWait is the default time allowance before an announced + // transaction is explicitly requested. + DefaultTxArrivalWait = 500 * time.Millisecond // txGatherSlack is the interval used to collate almost-expired announces // with network fetches. @@ -169,6 +169,8 @@ type TxFetcher struct { fetchTxs func(string, []common.Hash) error // Retrieves a set of txs from a remote peer dropPeer func(string) // Drops a peer in case of announcement violation + txArrivalWait time.Duration // Time allowance for an announced transaction to arrive before it is explicitly requested + step chan struct{} // Notification channel when the fetcher loop iterates clock mclock.Clock // Monotonic clock or simulated clock for tests realTime func() time.Time // Real system time or simulated time for tests @@ -176,16 +178,22 @@ type TxFetcher struct { } // NewTxFetcher creates a transaction fetcher to retrieve transaction -// based on hash announcements. -func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher { - return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil) +// based on hash announcements. txArrivalWait is how long an announced +// transaction may be waited on to arrive via broadcast before it is +// explicitly requested; zero requests it immediately. +func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), txArrivalWait time.Duration) *TxFetcher { + return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, txArrivalWait, mclock.System{}, time.Now, nil) } // NewTxFetcherForTests is a testing method to mock out the realtime clock with // a simulated version and the internal randomness with a deterministic one. func NewTxFetcherForTests( hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), - clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher { + txArrivalWait time.Duration, clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher { + if txArrivalWait < 0 { + txArrivalWait = 0 + } + return &TxFetcher{ notify: make(chan *txAnnounce), cleanup: make(chan *txDelivery), @@ -204,9 +212,12 @@ func NewTxFetcherForTests( addTxs: addTxs, fetchTxs: fetchTxs, dropPeer: dropPeer, - clock: clock, - realTime: realTime, - rand: rand, + + txArrivalWait: txArrivalWait, + + clock: clock, + realTime: realTime, + rand: rand, } } @@ -510,7 +521,7 @@ func (f *TxFetcher) loop() { actives := make(map[string]struct{}) for hash, instance := range f.waittime { - if time.Duration(f.clock.Now()-instance)+txGatherSlack > txArriveTimeout { + if time.Duration(f.clock.Now()-instance)+txGatherSlack > f.txArrivalWait { // Transaction expired without propagation, schedule for retrieval if f.announced[hash] != nil { panic("announce tracker already contains waitlist item") @@ -851,13 +862,13 @@ func (f *TxFetcher) rescheduleWait(timer *mclock.Timer, trigger chan struct{}) { for _, instance := range f.waittime { if earliest > instance { earliest = instance - if txArriveTimeout-time.Duration(now-earliest) < txGatherSlack { + if f.txArrivalWait-time.Duration(now-earliest) < txGatherSlack { break } } } - *timer = f.clock.AfterFunc(txArriveTimeout-time.Duration(now-earliest), func() { + *timer = f.clock.AfterFunc(f.txArrivalWait-time.Duration(now-earliest), func() { trigger <- struct{}{} }) } diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 40bbf9ce16..80e0227904 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/eth/fetcher/tx_fetcher_test.go @@ -102,6 +102,7 @@ func TestTransactionFetcherWaiting(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -187,7 +188,7 @@ func TestTransactionFetcherWaiting(t *testing.T) { // Wait for the arrival timeout which should move all expired items // from the wait list to the scheduler - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -225,7 +226,7 @@ func TestTransactionFetcherWaiting(t *testing.T) { {common.Hash{0x07}, types.LegacyTxType, 777}, }, }), - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isScheduled{ tracking: map[string][]announce{ "A": { @@ -304,6 +305,7 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -394,6 +396,7 @@ func TestTransactionFetcherSingletonRequesting(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -503,6 +506,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) { return errors.New("peer disconnected") }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -586,6 +590,7 @@ func TestTransactionFetcherCleanup(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -630,6 +635,7 @@ func TestTransactionFetcherCleanupEmpty(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -673,6 +679,7 @@ func TestTransactionFetcherMissingRescheduling(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -734,6 +741,7 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -783,14 +791,15 @@ func TestTransactionFetcherBroadcasts(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, 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())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[1]}, types: []byte{testTxs[1].Type()}, sizes: []uint32{uint32(testTxs[1].Size())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[2]}, types: []byte{testTxs[2].Type()}, sizes: []uint32{uint32(testTxs[2].Size())}}, isWaiting(map[string][]announce{ @@ -837,6 +846,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -847,7 +857,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) { }, }), isScheduled{nil, nil, nil}, - doWait{time: txArriveTimeout / 2, step: false}, + doWait{time: DefaultTxArrivalWait / 2, step: false}, isWaiting(map[string][]announce{ "A": { {common.Hash{0x01}, types.LegacyTxType, 111}, @@ -863,7 +873,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) { }, }), isScheduled{nil, nil, nil}, - doWait{time: txArriveTimeout / 2, step: true}, + doWait{time: DefaultTxArrivalWait / 2, step: true}, isWaiting(map[string][]announce{ "A": { {common.Hash{0x02}, types.LegacyTxType, 222}, @@ -909,6 +919,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -951,7 +962,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { types: []byte{testTxs[1].Type()}, sizes: []uint32{uint32(testTxs[1].Size())}, }, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isScheduled{ tracking: map[string][]announce{ "A": {{testTxsHashes[1], testTxs[1].Type(), uint32(testTxs[1].Size())}}, @@ -985,13 +996,14 @@ func TestTransactionFetcherTimeoutTimerResets(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "B", hashes: []common.Hash{{0x02}}, types: []byte{types.LegacyTxType}, sizes: []uint32{222}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ @@ -1063,13 +1075,14 @@ func TestTransactionFetcherRateLimiting(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ // Announce all the transactions, wait a bit and ensure only a small // percentage gets requested doTxNotify{peer: "A", hashes: hashes, types: ts, sizes: sizes}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -1093,6 +1106,7 @@ func TestTransactionFetcherBandwidthLimiting(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1116,7 +1130,7 @@ func TestTransactionFetcherBandwidthLimiting(t *testing.T) { types: []byte{types.BlobTxType, types.BlobTxType}, sizes: []uint32{params.BlobTxBlobGasPerBlob * 10, params.BlobTxBlobGasPerBlob * 10}, }, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -1192,13 +1206,14 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, 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]}, doTxNotify{peer: "B", hashes: hashesB[:maxTxAnnounces/2-1], types: typesB[:maxTxAnnounces/2-1], sizes: sizesB[:maxTxAnnounces/2-1]}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, // Announce the second half and keep them in the wait list doTxNotify{peer: "A", hashes: hashesA[maxTxAnnounces/2 : maxTxAnnounces], types: typesA[maxTxAnnounces/2 : maxTxAnnounces], sizes: sizesA[maxTxAnnounces/2 : maxTxAnnounces]}, @@ -1266,6 +1281,7 @@ func TestTransactionFetcherAnnouncementCapMatchesSenderQueue(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1301,6 +1317,7 @@ func TestTransactionFetcherFullCapDropsBatch(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1355,6 +1372,7 @@ func TestTransactionFetcherPartialBatchTrimming(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1408,6 +1426,7 @@ func TestTransactionFetcherPerPeerCapIndependence(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1470,12 +1489,13 @@ func TestTransactionFetcherCapSpansWaitAndScheduled(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ // Announce first half, let them move to scheduled doTxNotify{peer: "A", hashes: hashesFirst, types: typesFirst, sizes: sizesFirst}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), // Announce second half — should fill the remaining cap in waitlist doTxNotify{peer: "A", hashes: hashesSecond, types: typesSecond, sizes: sizesSecond}, @@ -1512,6 +1532,7 @@ func TestTransactionFetcherUnderpricedDedup(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1521,7 +1542,7 @@ func TestTransactionFetcherUnderpricedDedup(t *testing.T) { types: []byte{testTxs[0].Type(), testTxs[1].Type()}, sizes: []uint32{uint32(testTxs[0].Size()), uint32(testTxs[1].Size())}, }, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0], testTxs[1]}, direct: true}, isScheduled{nil, nil, nil}, @@ -1607,6 +1628,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: append(steps, []interface{}{ @@ -1617,7 +1639,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) { types: []byte{ts[maxTxUnderpricedSetSize]}, sizes: []uint32{sizes[maxTxUnderpricedSetSize]}, }, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{txs[maxTxUnderpricedSetSize]}, direct: true}, isUnderpriced(maxTxUnderpricedSetSize), }...), @@ -1635,6 +1657,7 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1647,9 +1670,9 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { // Set up a few hashes into various stages 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: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[1]}, types: []byte{testTxs[1].Type()}, sizes: []uint32{uint32(testTxs[1].Size())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[2]}, types: []byte{testTxs[2].Type()}, sizes: []uint32{uint32(testTxs[2].Size())}}, isWaiting(map[string][]announce{ @@ -1694,14 +1717,15 @@ func TestTransactionFetcherDrop(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ // Set up a few hashes into various stages doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{{0x02}}, types: []byte{types.LegacyTxType}, sizes: []uint32{222}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "A", hashes: []common.Hash{{0x03}}, types: []byte{types.LegacyTxType}, sizes: []uint32{333}}, isWaiting(map[string][]announce{ @@ -1727,7 +1751,7 @@ func TestTransactionFetcherDrop(t *testing.T) { // Push the node into a dangling (timeout) state 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: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -1768,12 +1792,13 @@ func TestTransactionFetcherDropRescheduling(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ // Set up a few hashes into various stages doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "B", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, isWaiting(nil), @@ -1814,6 +1839,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) { }, func(string, []common.Hash) error { return nil }, func(peer string) { drop <- peer }, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1856,7 +1882,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) { }, }), // Schedule all the transactions for retrieval - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -1897,12 +1923,13 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, 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())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}}, // Notify the dangling transaction once more and crash via a timeout @@ -1925,17 +1952,18 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, 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())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}}, // Notify the dangling transaction once more, re-fetch, and crash via a drop and timeout doTxNotify{peer: "B", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doDrop("A"), doWait{time: txFetchTimeout, step: true}, }, @@ -1955,6 +1983,7 @@ func TestTransactionFetcherFuzzCrash03(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -1970,7 +1999,7 @@ func TestTransactionFetcherFuzzCrash03(t *testing.T) { // Notify the dangling transaction once more, partially deliver, clash&crash with a timeout doTxNotify{peer: "B", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[1]}, direct: true}, doWait{time: txFetchTimeout, step: true}, @@ -1997,17 +2026,18 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) { return errors.New("peer disconnected") }, nil, + DefaultTxArrivalWait, ) }, 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())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}}, // Notify the dangling transaction once more, re-fetch, and crash via an in-flight disconnect doTxNotify{peer: "B", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, - doWait{time: txArriveTimeout, step: true}, + doWait{time: DefaultTxArrivalWait, step: true}, doFunc(func() { proceed <- struct{}{} // Allow peer A to return the failure }), @@ -2028,6 +2058,7 @@ func TestBlobTransactionAnnounce(t *testing.T) { nil, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -2067,7 +2098,7 @@ func TestBlobTransactionAnnounce(t *testing.T) { "B": {{0x03}}, }, }, - doWait{time: txArriveTimeout, step: true}, // zero time, but the blob fetching should be scheduled + doWait{time: DefaultTxArrivalWait, step: true}, // zero time, but the blob fetching should be scheduled isWaiting(nil), isScheduled{ tracking: map[string][]announce{ @@ -2098,11 +2129,12 @@ func TestTransactionFetcherDropAlternates(t *testing.T) { }, func(string, []common.Hash) error { return nil }, nil, + DefaultTxArrivalWait, ) }, 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}, + doWait{time: DefaultTxArrivalWait, step: true}, doTxNotify{peer: "B", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}}, isScheduled{ @@ -2187,6 +2219,7 @@ func TestTransactionProtocolViolation(t *testing.T) { return nil }, func(peer string) { drop <- struct{}{} }, + DefaultTxArrivalWait, ) }, steps: []interface{}{ @@ -2247,6 +2280,99 @@ func TestTransactionProtocolViolation(t *testing.T) { }) } +// Tests that a custom transaction arrival wait delays the explicit retrieval +// of announced transactions accordingly. +func TestTransactionFetcherCustomArrivalWait(t *testing.T) { + testTransactionFetcherParallel(t, txFetcherTest{ + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash) bool { return false }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + 250*time.Millisecond, + ) + }, + steps: []interface{}{ + doTxNotify{peer: "A", hashes: []common.Hash{{0x01}}, types: []byte{types.LegacyTxType}, sizes: []uint32{111}}, + isWaiting(map[string][]announce{ + "A": {{common.Hash{0x01}, types.LegacyTxType, 111}}, + }), + isScheduled{tracking: nil, fetching: nil}, + + // Half of the custom wait elapsed, the transaction should still be + // waiting for a potential broadcast + doWait{time: 125 * time.Millisecond, step: false}, + isWaiting(map[string][]announce{ + "A": {{common.Hash{0x01}, types.LegacyTxType, 111}}, + }), + isScheduled{tracking: nil, fetching: nil}, + + // Wait out the remainder, the transaction should be scheduled for + // retrieval + doWait{time: 125 * time.Millisecond, step: true}, + isWaiting(nil), + isScheduled{ + tracking: map[string][]announce{ + "A": {{common.Hash{0x01}, types.LegacyTxType, 111}}, + }, + fetching: map[string][]common.Hash{ + "A": {{0x01}}, + }, + }, + }, + }) +} + +// Tests that a zero transaction arrival wait skips the waitlist grace period +// and requests announced transactions immediately. +func TestTransactionFetcherZeroArrivalWait(t *testing.T) { + testTransactionFetcherParallel(t, txFetcherTest{ + init: func() *TxFetcher { + return NewTxFetcher( + func(common.Hash) bool { return false }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + 0, + ) + }, + steps: []interface{}{ + doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{types.LegacyTxType, types.LegacyTxType}, sizes: []uint32{111, 222}}, + doWait{time: 0, step: true}, + isWaiting(nil), + isScheduled{ + tracking: map[string][]announce{ + "A": { + {common.Hash{0x01}, types.LegacyTxType, 111}, + {common.Hash{0x02}, types.LegacyTxType, 222}, + }, + }, + fetching: map[string][]common.Hash{ + "A": {{0x01}, {0x02}}, + }, + }, + }, + }) +} + +// Tests that a negative transaction arrival wait is clamped to zero instead of +// breaking the waitlist timer arithmetic. +func TestTransactionFetcherNegativeArrivalWait(t *testing.T) { + t.Parallel() + + fetcher := NewTxFetcher( + func(common.Hash) bool { return false }, + nil, + func(string, []common.Hash) error { return nil }, + nil, + -time.Second, + ) + if fetcher.txArrivalWait != 0 { + t.Fatalf("negative arrival wait not clamped: have %v, want %v", fetcher.txArrivalWait, 0) + } +} + func testTransactionFetcherParallel(t *testing.T, tt txFetcherTest) { t.Parallel() testTransactionFetcher(t, tt) @@ -2622,6 +2748,7 @@ func TestTransactionForgotten(t *testing.T) { }, func(string, []common.Hash) error { return nil }, func(string) {}, + DefaultTxArrivalWait, mockClock, mockTime, rand.New(rand.NewSource(0)), // Use fixed seed for deterministic behavior diff --git a/eth/handler.go b/eth/handler.go index fc731a7579..26b165b797 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -129,6 +129,7 @@ type handlerConfig struct { EthAPI *ethapi.BlockChainAPI // EthAPI to interact enableBlockTracking bool // Whether to log information collected while tracking block lifecycle txAnnouncementOnly bool // Whether to only announce txs to peers + txArrivalWait time.Duration // Time allowance for an announced tx to arrive before explicitly requesting it disableTxPropagation bool // Whether to disable broadcasting and announcement of txs to peers witnessProtocol bool // Whether to enable witness protocol syncWithWitnesses bool // Whether to sync blocks with witnesses @@ -319,7 +320,7 @@ func newHandler(config *handlerConfig) (*handler, error) { addTxs := func(txs []*types.Transaction) []error { return h.txpool.Add(txs, false) } - h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, addTxs, fetchTx, h.removePeer) + h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, addTxs, fetchTx, h.removePeer, config.txArrivalWait) h.chainSync = newChainSyncer(h) return h, nil diff --git a/eth/handler_test.go b/eth/handler_test.go index 9715f06bf2..aa80cc7c31 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -37,6 +37,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth/downloader" + "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" @@ -214,12 +215,13 @@ func newTestHandlerWithBlocks(blocks int) *testHandler { txpool := newTestTxPool() handler, _ := newHandler(&handlerConfig{ - Database: db, - Chain: chain, - TxPool: txpool, - Network: 1, - Sync: downloader.SnapSync, - BloomCache: 1, + Database: db, + Chain: chain, + TxPool: txpool, + Network: 1, + Sync: downloader.SnapSync, + BloomCache: 1, + txArrivalWait: fetcher.DefaultTxArrivalWait, }) handler.Start(1000) diff --git a/internal/cli/server/config.go b/internal/cli/server/config.go index 671a47438f..de6a042d3d 100644 --- a/internal/cli/server/config.go +++ b/internal/cli/server/config.go @@ -29,6 +29,7 @@ import ( "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/eth/downloader/whitelist" "github.com/ethereum/go-ethereum/eth/ethconfig" + "github.com/ethereum/go-ethereum/eth/fetcher" "github.com/ethereum/go-ethereum/eth/gasprice" "github.com/ethereum/go-ethereum/internal/cli/server/chains" "github.com/ethereum/go-ethereum/log" @@ -859,7 +860,7 @@ func DefaultConfig() *Config { NoDiscover: false, NAT: "any", NetRestrict: "", - TxArrivalWait: 500 * time.Millisecond, + TxArrivalWait: fetcher.DefaultTxArrivalWait, TxAnnouncementOnly: false, DisableTxPropagation: false, NoSnapServing: false, diff --git a/internal/cli/server/config_test.go b/internal/cli/server/config_test.go index 45c9e86faa..673d750b44 100644 --- a/internal/cli/server/config_test.go +++ b/internal/cli/server/config_test.go @@ -30,6 +30,25 @@ func assertBorDefaultGasPrice(t *testing.T, ethConfig *ethconfig.Config) { assert.Equal(t, ethConfig.Miner.GasPrice, big.NewInt(params.BorDefaultMinerGasPrice)) } +func TestTxArrivalWaitConfig(t *testing.T) { + t.Parallel() + + config := DefaultConfig() + assert.NoError(t, config.loadChain()) + + nodeCfg, err := config.buildNode() + assert.NoError(t, err) + assert.Equal(t, 500*time.Millisecond, nodeCfg.P2P.TxArrivalWait) + + config.P2P.TxArrivalWaitRaw = "0s" + assert.NoError(t, config.fillTimeDurations()) + assert.Equal(t, time.Duration(0), config.P2P.TxArrivalWait) + + nodeCfg, err = config.buildNode() + assert.NoError(t, err) + assert.Equal(t, time.Duration(0), nodeCfg.P2P.TxArrivalWait) +} + func TestConfigMerge(t *testing.T) { c0 := &Config{ Chain: "0", diff --git a/internal/cli/server/flags.go b/internal/cli/server/flags.go index c7e5e4b128..cb7b91dbd5 100644 --- a/internal/cli/server/flags.go +++ b/internal/cli/server/flags.go @@ -1060,7 +1060,7 @@ func (c *Command) Flags(config *Config) *flagset.Flagset { }) f.DurationFlag(&flagset.DurationFlag{ Name: "txarrivalwait", - Usage: "Maximum duration to wait for a transaction before explicitly requesting it", + Usage: "Maximum duration to wait for a transaction before explicitly requesting it (0 requests announced transactions immediately)", Value: &c.cliConfig.P2P.TxArrivalWait, Default: c.cliConfig.P2P.TxArrivalWait, Group: "P2P", diff --git a/p2p/config.go b/p2p/config.go index d7e29b8423..9de5f7d4c8 100644 --- a/p2p/config.go +++ b/p2p/config.go @@ -127,8 +127,10 @@ type Config struct { clock mclock.Clock - // TxArrivalWait is the duration (ms) that the node will wait after seeing - // an announced transaction before explicitly requesting it + // TxArrivalWait is the duration that the node will wait after seeing an + // announced transaction before explicitly requesting it, in case the + // transaction arrives via broadcast in the meantime. A value of zero + // requests announced transactions immediately. TxArrivalWait time.Duration // TxAnnouncementOnly is used to only announce transactions to peers diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index 0fa2e6d503..56589d2228 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go @@ -87,6 +87,7 @@ func fuzz(input []byte) int { }, func(string, []common.Hash) error { return nil }, nil, + fetcher.DefaultTxArrivalWait, clock, func() time.Time { nanoTime := int64(clock.Now())