feat: fire and forget

Author: minhd-vu

cmd/p2p/sensor/rpc.go

@@ -138,11 +138,14 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 
 		params.counter.WithLabelValues(req.Method, "false").Inc()
 
-		// Broadcast any transactions (always broadcast RPC-submitted transactions)
+		// Broadcast transactions asynchronously to avoid blocking the RPC response
 		if len(txs) > 0 {
-			log.Info().Str("hash", txs[0].Hash().Hex()).Msg("Broadcasting transaction")
-			count := params.conns.BroadcastTxsAlways(txs)
-			log.Info().Str("hash", txs[0].Hash().Hex()).Int("peers", count).Msg("Transaction broadcast complete")
+			go func(txs types.Transactions) {
+				hash := txs[0].Hash().Hex()
+				log.Info().Str("hash", hash).Msg("Broadcasting transaction")
+				count := params.conns.BroadcastTxsAlways(txs)
+				log.Info().Str("hash", hash).Int("peers", count).Msg("Transaction broadcast complete")
+			}(txs)
 		}
 
 		// Write response
@@ -288,18 +291,13 @@ func handleBatchRequest(w http.ResponseWriter, r *http.Request, body []byte, par
 		}
 	}
 
+	// Broadcast transactions asynchronously to avoid blocking the RPC response
 	if len(txs) > 0 {
-		log.Info().
-			Int("txs", len(txs)).
-			Int("requests", len(requests)).
-			Msg("Broadcasting batch of transactions")
-
-		count := params.conns.BroadcastTxsAlways(txs)
-
-		log.Info().
-			Int("txs", len(txs)).
-			Int("peers", count).
-			Msg("Batch broadcast complete")
+		go func(txs types.Transactions) {
+			log.Info().Int("txs", len(txs)).Msg("Broadcasting batch of transactions")
+			count := params.conns.BroadcastTxsAlways(txs)
+			log.Info().Int("txs", len(txs)).Int("peers", count).Msg("Batch broadcast complete")
+		}(txs)
 	}
 
 	if len(indices) > 0 {

loadtest/preconf.go

@@ -102,7 +102,6 @@ func (pt *PreconfTracker) Track(txHash common.Hash) {
 	var preconfError error
 	var preconfDuration time.Duration
 	wg.Go(func() {
-
 		preconfStartTime := time.Now()
 		defer func() {
 			preconfDuration = time.Since(preconfStartTime)
@@ -116,7 +115,6 @@ func (pt *PreconfTracker) Track(txHash common.Hash) {
 	var receiptError error
 	var receiptDuration time.Duration
 	wg.Go(func() {
-
 		time.Sleep(100 * time.Millisecond)
 
 		receiptTime := time.Now()

p2p/conns.go

@@ -154,13 +154,10 @@ func (c *Conns) BroadcastTxs(txs types.Transactions) int {
 	var count atomic.Int32
 	var wg sync.WaitGroup
 
-	for _, cn := range peers {
-		wg.Add(1)
-		go func(cn *conn) {
-			defer wg.Done()
-
+	for _, peer := range peers {
+		wg.Go(func() {
 			// Filter transactions this peer doesn't know about using batch bloom operation
-			unknownHashes := cn.filterUnknownTxHashes(hashes)
+			unknownHashes := peer.filterUnknownTxHashes(hashes)
 			if len(unknownHashes) == 0 {
 				return
 			}
@@ -172,32 +169,25 @@ func (c *Conns) BroadcastTxs(txs types.Transactions) int {
 					unknownTxs = append(unknownTxs, tx)
 				}
 			}
-
 			if len(unknownTxs) == 0 {
 				return
 			}
 
-			// Send as TransactionsPacket
 			rawList, err := rlp.EncodeToRawList([]*types.Transaction(unknownTxs))
 			if err != nil {
-				cn.logger.Debug().
-					Err(err).
-					Msg("Failed to encode transactions")
+				peer.logger.Debug().Err(err).Msg("Failed to encode transactions")
 				return
 			}
 			packet := &eth.TransactionsPacket{RawList: rawList}
-			cn.countMsgSent(packet.Name(), float64(len(unknownTxs)))
-			if err := ethp2p.Send(cn.rw, eth.TransactionsMsg, packet); err != nil {
-				cn.logger.Debug().
-					Err(err).
-					Msg("Failed to send transactions")
+			peer.countMsgSent(packet.Name(), float64(len(unknownTxs)))
+			if err := ethp2p.Send(peer.rw, eth.TransactionsMsg, packet); err != nil {
+				peer.logger.Debug().Err(err).Msg("Failed to send transactions")
 				return
 			}
 
-			// Mark transactions as known for this peer using batch bloom operation
-			cn.addKnownTxHashes(unknownHashes)
+			peer.addKnownTxHashes(unknownHashes)
 			count.Add(1)
-		}(cn)
+		})
 	}
 
 	wg.Wait()
@@ -213,17 +203,13 @@ func (c *Conns) BroadcastTxs(txs types.Transactions) int {
 	return finalCount
 }
 
-// BroadcastTxsAlways broadcasts transactions to all connected peers regardless
-// of the ShouldBroadcastTx configuration. This is used for transactions
-// submitted via RPC that must always be broadcast to the network.
-// Unlike BroadcastTxs, this does not filter by known transactions since
-// RPC-submitted transactions are always new.
+// BroadcastTxsAlways broadcasts RPC-submitted transactions to all peers unconditionally.
+// Always uses TransactionsMsg (matching Bor behavior - PooledTransactionsMsg is response-only).
 func (c *Conns) BroadcastTxsAlways(txs types.Transactions) int {
 	if len(txs) == 0 {
 		return 0
 	}
 
-	// Pre-compute transaction hashes once to avoid redundant Keccak256 computations
 	hashes := make([]common.Hash, len(txs))
 	for i, tx := range txs {
 		hashes[i] = tx.Hash()
@@ -234,51 +220,43 @@ func (c *Conns) BroadcastTxsAlways(txs types.Transactions) int {
 		return 0
 	}
 
-	// Broadcast concurrently to all peers
+	return c.broadcastTxs(txs, hashes, peers)
+}
+
+// broadcastTxs sends transactions directly via TransactionsMsg.
+func (c *Conns) broadcastTxs(txs types.Transactions, hashes []common.Hash, peers []*conn) int {
 	var count atomic.Int32
 	var wg sync.WaitGroup
 
-	for _, cn := range peers {
-		wg.Add(1)
-		go func(cn *conn) {
-			defer wg.Done()
-
-			// Send as TransactionsPacket
+	for _, peer := range peers {
+		wg.Go(func() {
 			rawList, err := rlp.EncodeToRawList([]*types.Transaction(txs))
 			if err != nil {
-				cn.logger.Debug().
-					Err(err).
-					Msg("Failed to encode transactions")
+				peer.logger.Debug().Err(err).Msg("Failed to encode transactions")
 				return
 			}
 			packet := &eth.TransactionsPacket{RawList: rawList}
-			cn.countMsgSent(packet.Name(), float64(len(txs)))
-			if err := ethp2p.Send(cn.rw, eth.TransactionsMsg, packet); err != nil {
-				cn.logger.Debug().
-					Err(err).
-					Msg("Failed to send transactions")
+			peer.countMsgSent(packet.Name(), float64(len(txs)))
+			if err := ethp2p.Send(peer.rw, eth.TransactionsMsg, packet); err != nil {
+				peer.logger.Debug().Err(err).Msg("Failed to send transactions")
 				return
 			}
-
-			// Mark transactions as known for this peer using batch bloom operation
-			cn.addKnownTxHashes(hashes)
+			peer.addKnownTxHashes(hashes)
 			count.Add(1)
-		}(cn)
+		})
 	}
 
 	wg.Wait()
 
-	finalCount := int(count.Load())
-	if finalCount > 0 {
-		log.Debug().
-			Int("peers", finalCount).
-			Int("txs", len(txs)).
-			Msg("Broadcasted transactions (always)")
+	pc := int(count.Load())
+	if pc > 0 {
+		log.Debug().Int("peers", pc).Int("txs", len(txs)).Msg("Broadcasted transactions")
 	}
 
-	return finalCount
+	return pc
 }
 
+
 // BroadcastTxHashes enqueues transaction hashes to per-peer broadcast queues.
 // Each peer has a dedicated goroutine that drains the queue and batches sends.
 // Returns the number of peers the hashes were enqueued to.
@@ -333,35 +311,22 @@ func (c *Conns) BroadcastBlock(block *types.Block, td *big.Int) int {
 	var count atomic.Int32
 	var wg sync.WaitGroup
 
-	for _, cn := range peers {
-		wg.Add(1)
-		go func(cn *conn) {
-			defer wg.Done()
-
-			// Skip if peer already knows about this block
-			if cn.hasKnownBlock(hash) {
+	for _, peer := range peers {
+		wg.Go(func() {
+			if peer.hasKnownBlock(hash) {
 				return
 			}
 
-			// Send NewBlockPacket
-			packet := eth.NewBlockPacket{
-				Block: block,
-				TD:    td,
-			}
-
-			cn.countMsgSent(packet.Name(), 1)
-			if err := ethp2p.Send(cn.rw, eth.NewBlockMsg, &packet); err != nil {
-				cn.logger.Debug().
-					Err(err).
-					Uint64("number", block.Number().Uint64()).
-					Msg("Failed to send block")
+			packet := eth.NewBlockPacket{Block: block, TD: td}
+			peer.countMsgSent(packet.Name(), 1)
+			if err := ethp2p.Send(peer.rw, eth.NewBlockMsg, &packet); err != nil {
+				peer.logger.Debug().Err(err).Uint64("number", block.Number().Uint64()).Msg("Failed to send block")
 				return
 			}
 
-			// Mark block as known for this peer
-			cn.addKnownBlock(hash)
+			peer.addKnownBlock(hash)
 			count.Add(1)
-		}(cn)
+		})
 	}
 
 	wg.Wait()