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