fix: max out blocks

Author: minhd-vu

cmd/p2p/sensor/rpc.go

@@ -138,16 +138,9 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 
 		params.counter.WithLabelValues(req.Method, "false").Inc()
 
-		// Broadcast transactions asynchronously to avoid blocking the RPC response
+		// Enqueue transactions for async broadcast
 		if len(txs) > 0 {
-			go func(txs types.Transactions) {
-				hash := txs[0].Hash().Hex()
-				count := params.conns.BroadcastTxsAlways(txs)
-				log.Debug().
-					Str("hash", hash).
-					Int("peers", count).
-					Msg("Broadcasted transaction")
-			}(txs)
+			params.conns.EnqueueTxBroadcast(txs)
 		}
 
 		// Write response
@@ -293,15 +286,9 @@ func handleBatchRequest(w http.ResponseWriter, r *http.Request, body []byte, par
 		}
 	}
 
-	// Broadcast transactions asynchronously to avoid blocking the RPC response
+	// Enqueue transactions for async broadcast
 	if len(txs) > 0 {
-		go func(txs types.Transactions) {
-			count := params.conns.BroadcastTxsAlways(txs)
-			log.Debug().
-				Int("txs", len(txs)).
-				Int("peers", count).
-				Msg("Broadcasted transaction batch")
-		}(txs)
+		params.conns.EnqueueTxBroadcast(txs)
 	}
 
 	if len(indices) > 0 {

p2p/conns.go

@@ -67,6 +67,13 @@ type Conns struct {
 	shouldBroadcastTxHashes    bool
 	shouldBroadcastBlocks      bool
 	shouldBroadcastBlockHashes bool
+
+	// txBroadcastQueue is an unbounded queue of transactions for async broadcast
+	txBroadcastQueue []types.Transactions
+	txBroadcastMu    sync.Mutex
+	txBroadcastCond  *sync.Cond
+	// closeCh signals the broadcast worker to stop
+	closeCh chan struct{}
 }
 
 // NewConns creates a new connection manager with a blocks cache.
@@ -77,7 +84,7 @@ func NewConns(opts ConnsOptions) *Conns {
 	oldest := &ds.Locked[*types.Header]{}
 	oldest.Set(opts.Head.Block.Header())
 
-	return &Conns{
+	c := &Conns{
 		conns:                      make(map[string]*conn),
 		blocks:                     ds.NewLRU[common.Hash, BlockCache](opts.BlocksCache),
 		txs:                        ds.NewLRU[common.Hash, *types.Transaction](opts.TxsCache),
@@ -89,7 +96,11 @@ func NewConns(opts ConnsOptions) *Conns {
 		shouldBroadcastTxHashes:    opts.ShouldBroadcastTxHashes,
 		shouldBroadcastBlocks:      opts.ShouldBroadcastBlocks,
 		shouldBroadcastBlockHashes: opts.ShouldBroadcastBlockHashes,
+		closeCh:                    make(chan struct{}),
 	}
+	c.txBroadcastCond = sync.NewCond(&c.txBroadcastMu)
+	go c.txBroadcastLoop()
+	return c
 }
 
 // Add adds a connection to the manager.
@@ -256,6 +267,99 @@ func (c *Conns) broadcastTxs(txs types.Transactions, hashes []common.Hash, peers
 	return pc
 }
 
+// txBroadcastLoop is the worker that drains the broadcast queue and sends
+// transactions to all peers, pulling only up to max packet size per iteration.
+// While broadcasting, more transactions accumulate in the queue for the next batch.
+func (c *Conns) txBroadcastLoop() {
+	for {
+		batch, done := c.pullTxBatch()
+		if done {
+			return
+		}
+		if len(batch) == 0 {
+			continue
+		}
+
+		peers := c.snapshotPeers()
+		if len(peers) == 0 {
+			continue
+		}
+
+		hashes := make([]common.Hash, len(batch))
+		for i, tx := range batch {
+			hashes[i] = tx.Hash()
+		}
+
+		c.broadcastTxs(batch, hashes, peers)
+		log.Info().Int("txs", len(batch)).Int("peers", len(peers)).Msg("Broadcasted batch")
+	}
+}
+
+// pullTxBatch pulls transactions from the queue up to maxTxPacketSize.
+// Returns the batch and whether the worker should exit.
+func (c *Conns) pullTxBatch() (types.Transactions, bool) {
+	c.txBroadcastMu.Lock()
+	defer c.txBroadcastMu.Unlock()
+
+	// Wait for transactions or close signal
+	for len(c.txBroadcastQueue) == 0 {
+		select {
+		case <-c.closeCh:
+			return nil, true
+		default:
+		}
+		c.txBroadcastCond.Wait()
+	}
+
+	var batch types.Transactions
+	var batchSize uint64
+
+	// Pull from queue until we hit max packet size
+	for len(c.txBroadcastQueue) > 0 {
+		txs := c.txBroadcastQueue[0]
+
+		// Try to fit transactions from this slice
+		for len(txs) > 0 {
+			tx := txs[0]
+			txSize := tx.Size()
+
+			// If adding this tx would exceed max size and we have something, stop
+			if len(batch) > 0 && batchSize+txSize > maxTxPacketSize {
+				// Put remaining txs back at front of queue
+				if len(txs) > 0 {
+					c.txBroadcastQueue[0] = txs
+				} else {
+					c.txBroadcastQueue = c.txBroadcastQueue[1:]
+				}
+				return batch, false
+			}
+
+			batch = append(batch, tx)
+			batchSize += txSize
+			txs = txs[1:]
+		}
+
+		// Consumed entire slice, remove from queue
+		c.txBroadcastQueue = c.txBroadcastQueue[1:]
+	}
+
+	return batch, false
+}
+
+// EnqueueTxBroadcast adds transactions to the broadcast queue for async sending.
+// The queue is unbounded; transactions are never dropped.
+func (c *Conns) EnqueueTxBroadcast(txs types.Transactions) {
+	c.txBroadcastMu.Lock()
+	c.txBroadcastQueue = append(c.txBroadcastQueue, txs)
+	c.txBroadcastMu.Unlock()
+	c.txBroadcastCond.Signal()
+}
+
+// Close stops the broadcast worker.
+func (c *Conns) Close() {
+	close(c.closeCh)
+	c.txBroadcastCond.Broadcast()
+}
 
 // BroadcastTxHashes enqueues transaction hashes to per-peer broadcast queues.
 // Each peer has a dedicated goroutine that drains the queue and batches sends.