eth: avoid transaction pool query when annoucing transaction

With eth/68, transaction announcement must have transaction type and size. So
in announceTransactions, we need to query the transaction from transaction pool
with its hash. This creates overhead in case of blob transaction which needs to
load data from limbo and RLP decode. This commit tries to avoid transaction
pool query by providing the transaction type and size information along with
hash right from the beginning.

core/types/transaction.go

@@ -56,6 +56,12 @@ type Transaction struct {
 	inner TxData    // Consensus contents of a transaction
 	time  time.Time // Time first seen locally (spam avoidance)
 
+	// This is only set in blob transaction without sidecar
+	// so that we can quickly get the size of whole blob
+	// transaction without needing to query to database to get
+	// back the whole blob transaction with sidecar
+	sidecarSize uint64
+
 	// caches
 	hash atomic.Pointer[common.Hash]
 	size atomic.Uint64
@@ -444,6 +450,9 @@ func (tx *Transaction) WithoutBlobTxSidecar() *Transaction {
 		inner: blobtx.withoutSidecar(),
 		time:  tx.time,
 	}
+	if sc := blobtx.Sidecar; sc != nil {
+		cpy.sidecarSize = rlp.ListSize(sc.encodedSize())
+	}
 	// Note: tx.size cache not carried over because the sidecar is included in size!
 	if h := tx.hash.Load(); h != nil {
 		cpy.hash.Store(h)
@@ -563,6 +572,10 @@ func (tx *Transaction) Size() uint64 {
 	return size
 }
 
+func (tx *Transaction) SidecarSize() uint64 {
+	return tx.sidecarSize
+}
+
 // WithSignature returns a new transaction with the given signature.
 // This signature needs to be in the [R || S || V] format where V is 0 or 1.
 func (tx *Transaction) WithSignature(signer Signer, sig []byte) (*Transaction, error) {

eth/handler.go

@@ -465,8 +465,8 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		directCount int // Number of transactions sent directly to peers (duplicates included)
 		annCount    int // Number of transactions announced across all peers (duplicates included)
 
-		txset = make(map[*ethPeer][]common.Hash) // Set peer->hash to transfer directly
-		annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce
+		txset = make(map[*ethPeer][]common.Hash)         // Set peer->hash to transfer directly
+		annos = make(map[*ethPeer][]*eth.TxAnnouncement) // Set peer->txAnnouncement to announce
 	)
 	// Broadcast transactions to a batch of peers not knowing about it
 	direct := big.NewInt(int64(math.Sqrt(float64(h.peers.len())))) // Approximate number of peers to broadcast to
@@ -515,7 +515,14 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 			if broadcast {
 				txset[peer] = append(txset[peer], tx.Hash())
 			} else {
-				annos[peer] = append(annos[peer], tx.Hash())
+				annos[peer] = append(annos[peer], &eth.TxAnnouncement{
+					Hash: tx.Hash(),
+					Type: tx.Type(),
+					// If this is a blob transaction, tx is already a
+					// transaction without sidecar. So we need to add
+					// the sidecar size to get the correct size.
+					Size: tx.Size() + tx.SidecarSize(),
+				})
 			}
 		}
 	}
@@ -523,9 +530,9 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		directCount += len(hashes)
 		peer.AsyncSendTransactions(hashes)
 	}
-	for peer, hashes := range annos {
-		annCount += len(hashes)
-		peer.AsyncSendPooledTransactionHashes(hashes)
+	for peer, announces := range annos {
+		annCount += len(announces)
+		peer.AsyncSendPooledTransactionHashes(announces)
 	}
 	log.Debug("Distributed transactions", "plaintxs", len(txs)-blobTxs-largeTxs, "blobtxs", blobTxs, "largetxs", largeTxs,
 		"bcastpeers", len(txset), "bcastcount", directCount, "annpeers", len(annos), "anncount", annCount)

eth/protocols/eth/broadcast.go

@@ -27,6 +27,12 @@ const (
 	maxTxPacketSize = 100 * 1024
 )
 
+type TxAnnouncement struct {
+	Hash common.Hash
+	Type uint8
+	Size uint64
+}
+
 // broadcastTransactions is a write loop that schedules transaction broadcasts
 // to the remote peer. The goal is to have an async writer that does not lock up
 // node internals and at the same time rate limits queued data.
@@ -99,7 +105,7 @@ func (p *Peer) broadcastTransactions() {
 // node internals and at the same time rate limits queued data.
 func (p *Peer) announceTransactions() {
 	var (
-		queue  []common.Hash         // Queue of hashes to announce as transaction stubs
+		queue  []*TxAnnouncement     // Queue of hashes to announce as transaction stubs
 		done   chan struct{}         // Non-nil if background announcer is running
 		fail   = make(chan error, 1) // Channel used to receive network error
 		failed bool                  // Flag whether a send failed, discard everything onward
@@ -116,12 +122,12 @@ func (p *Peer) announceTransactions() {
 				size         common.StorageSize
 			)
 			for count = 0; count < len(queue) && size < maxTxPacketSize; count++ {
-				if tx := p.txpool.Get(queue[count]); tx != nil {
-					pending = append(pending, queue[count])
-					pendingTypes = append(pendingTypes, tx.Type())
-					pendingSizes = append(pendingSizes, uint32(tx.Size()))
-					size += common.HashLength
-				}
+				announce := queue[count]
+
+				pending = append(pending, announce.Hash)
+				pendingTypes = append(pendingTypes, announce.Type)
+				pendingSizes = append(pendingSizes, uint32(announce.Size))
+				size += common.HashLength
 			}
 			// Shift and trim queue
 			queue = queue[:copy(queue, queue[count:])]
@@ -141,13 +147,13 @@ func (p *Peer) announceTransactions() {
 		}
 		// Transfer goroutine may or may not have been started, listen for events
 		select {
-		case hashes := <-p.txAnnounce:
+		case announces := <-p.txAnnounce:
 			// If the connection failed, discard all transaction events
 			if failed {
 				continue
 			}
 			// New batch of transactions to be broadcast, queue them (with cap)
-			queue = append(queue, hashes...)
+			queue = append(queue, announces...)
 			if len(queue) > maxQueuedTxAnns {
 				// Fancy copy and resize to ensure buffer doesn't grow indefinitely
 				queue = queue[:copy(queue, queue[len(queue)-maxQueuedTxAnns:])]

eth/protocols/eth/peer.go

@@ -48,10 +48,10 @@ type Peer struct {
 	rw        p2p.MsgReadWriter // Input/output streams for snap
 	version   uint              // Protocol version negotiated
 
-	txpool      TxPool             // Transaction pool used by the broadcasters for liveness checks
-	knownTxs    *knownCache        // Set of transaction hashes known to be known by this peer
-	txBroadcast chan []common.Hash // Channel used to queue transaction propagation requests
-	txAnnounce  chan []common.Hash // Channel used to queue transaction announcement requests
+	txpool      TxPool                 // Transaction pool used by the broadcasters for liveness checks
+	knownTxs    *knownCache            // Set of transaction hashes known to be known by this peer
+	txBroadcast chan []common.Hash     // Channel used to queue transaction propagation requests
+	txAnnounce  chan []*TxAnnouncement // Channel used to queue transaction announcement requests
 
 	reqDispatch chan *request  // Dispatch channel to send requests and track then until fulfillment
 	reqCancel   chan *cancel   // Dispatch channel to cancel pending requests and untrack them
@@ -70,7 +70,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe
 		version:     version,
 		knownTxs:    newKnownCache(maxKnownTxs),
 		txBroadcast: make(chan []common.Hash),
-		txAnnounce:  make(chan []common.Hash),
+		txAnnounce:  make(chan []*TxAnnouncement),
 		reqDispatch: make(chan *request),
 		reqCancel:   make(chan *cancel),
 		resDispatch: make(chan *response),
@@ -160,13 +160,15 @@ func (p *Peer) sendPooledTransactionHashes(hashes []common.Hash, types []byte, s
 // AsyncSendPooledTransactionHashes queues a list of transactions hashes to eventually
 // announce to a remote peer.  The number of pending sends are capped (new ones
 // will force old sends to be dropped)
-func (p *Peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) {
+func (p *Peer) AsyncSendPooledTransactionHashes(txAnnounce []*TxAnnouncement) {
 	select {
-	case p.txAnnounce <- hashes:
+	case p.txAnnounce <- txAnnounce:
 		// Mark all the transactions as known, but ensure we don't overflow our limits
-		p.knownTxs.Add(hashes...)
+		for _, announce := range txAnnounce {
+			p.knownTxs.Add(announce.Hash)
+		}
 	case <-p.term:
-		p.Log().Debug("Dropping transaction announcement", "count", len(hashes))
+		p.Log().Debug("Dropping transaction announcement", "count", len(txAnnounce))
 	}
 }
 

eth/sync.go

@@ -17,21 +17,26 @@
 package eth
 
 import (
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 )
 
 // syncTransactions starts sending all currently pending transactions to the given peer.
 func (h *handler) syncTransactions(p *eth.Peer) {
-	var hashes []common.Hash
+	var announces []*eth.TxAnnouncement
 	for _, batch := range h.txpool.Pending(txpool.PendingFilter{OnlyPlainTxs: true}) {
 		for _, tx := range batch {
-			hashes = append(hashes, tx.Hash)
+			resolved := tx.Resolve()
+
+			announces = append(announces, &eth.TxAnnouncement{
+				Hash: tx.Hash,
+				Size: resolved.Size(),
+				Type: resolved.Type(),
+			})
 		}
 	}
-	if len(hashes) == 0 {
+	if len(announces) == 0 {
 		return
 	}
-	p.AsyncSendPooledTransactionHashes(hashes)
+	p.AsyncSendPooledTransactionHashes(announces)
 }