feat: add configurable tx batch timeout with 500ms default

Author: minhd-vu

cmd/p2p/sensor/sensor.go

@@ -58,6 +58,7 @@ type (
 		ShouldBroadcastBlocks        bool
 		ShouldBroadcastBlockHashes   bool
 		BroadcastWorkers             int
+		TxBatchTimeout               time.Duration
 		ShouldRunPprof               bool
 		PprofPort                    uint
 		ShouldRunPrometheus          bool
@@ -219,6 +220,7 @@ var SensorCmd = &cobra.Command{
 			ShouldBroadcastBlocks:      inputSensorParams.ShouldBroadcastBlocks,
 			ShouldBroadcastBlockHashes: inputSensorParams.ShouldBroadcastBlockHashes,
 			BroadcastWorkers:           inputSensorParams.BroadcastWorkers,
+			TxBatchTimeout:             inputSensorParams.TxBatchTimeout,
 		})
 
 		opts := p2p.EthProtocolOptions{
@@ -494,6 +496,7 @@ will result in less chance of missing data but can significantly increase memory
 	f.BoolVar(&inputSensorParams.ShouldBroadcastBlocks, "broadcast-blocks", false, "broadcast full blocks to peers")
 	f.BoolVar(&inputSensorParams.ShouldBroadcastBlockHashes, "broadcast-block-hashes", false, "broadcast block hashes to peers")
 	f.IntVar(&inputSensorParams.BroadcastWorkers, "broadcast-workers", 4, "number of concurrent broadcast workers")
+	f.DurationVar(&inputSensorParams.TxBatchTimeout, "tx-batch-timeout", 500*time.Millisecond, "timeout for batching transactions before broadcast")
 	f.BoolVar(&inputSensorParams.ShouldRunPprof, "pprof", false, "run pprof server")
 	f.UintVar(&inputSensorParams.PprofPort, "pprof-port", 6060, "port pprof runs on")
 	f.BoolVar(&inputSensorParams.ShouldRunPrometheus, "prom", true, "run Prometheus server")

p2p/conns.go

@@ -37,6 +37,7 @@ type ConnsOptions struct {
 	ShouldBroadcastBlocks      bool
 	ShouldBroadcastBlockHashes bool
 	BroadcastWorkers           int
+	TxBatchTimeout             time.Duration
 }
 
 // Conns manages a collection of active peer connections for transaction broadcasting.
@@ -72,6 +73,9 @@ type Conns struct {
 
 	// txBroadcastCh is a buffered channel for async transaction broadcast
 	txBroadcastCh chan types.Transactions
+
+	// txBatchTimeout is the timeout for batching transactions before broadcast
+	txBatchTimeout time.Duration
 }
 
 // NewConns creates a new connection manager with a blocks cache.
@@ -82,6 +86,11 @@ func NewConns(opts ConnsOptions) *Conns {
 	oldest := &ds.Locked[*types.Header]{}
 	oldest.Set(opts.Head.Block.Header())
 
+	txBatchTimeout := opts.TxBatchTimeout
+	if txBatchTimeout <= 0 {
+		txBatchTimeout = 500 * time.Millisecond
+	}
+
 	c := &Conns{
 		conns:                      make(map[string]*conn),
 		blocks:                     ds.NewLRU[common.Hash, BlockCache](opts.BlocksCache),
@@ -95,6 +104,7 @@ func NewConns(opts ConnsOptions) *Conns {
 		shouldBroadcastBlocks:      opts.ShouldBroadcastBlocks,
 		shouldBroadcastBlockHashes: opts.ShouldBroadcastBlockHashes,
 		txBroadcastCh:              make(chan types.Transactions, 100000),
+		txBatchTimeout:             txBatchTimeout,
 	}
 
 	workers := opts.BroadcastWorkers
@@ -311,7 +321,7 @@ func (c *Conns) pullTxBatch() (types.Transactions, []common.Hash) {
 	}
 
 	// Drain more until max size or timeout
-	timer := time.NewTimer(100 * time.Millisecond)
+	timer := time.NewTimer(c.txBatchTimeout)
 	defer timer.Stop()
 
 	for batchSize < maxTxPacketSize {