eth, eth/fetcher, p2p: make tx arrival wait configurable via p2p.txarrivalwait

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)
 

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,

eth/fetcher/tx_fetcher.go

@@ -64,9 +64,9 @@
 	// 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,23 +169,31 @@
 	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
 	rand     *mrand.Rand      // Randomizer to use in tests instead of map range loops (soft-random)
 }
 
 // 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 @@
 		addTxs:      addTxs,
 		fetchTxs:    fetchTxs,
 		dropPeer:    dropPeer,
-		clock:       clock,
-		realTime:    realTime,
-		rand:        rand,
+
+		txArrivalWait: txArrivalWait,
+
+		clock:    clock,
+		realTime: realTime,
+		rand:     rand,
 	}
 }
 
@@ -510,7 +521,7 @@
 			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 @@
 	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{}{}
 	})
 }