core/txpool/blobpool: speed up serving pre eth/72 peers (#35543)

Add cache for recovered cells.

Serving a blob tx to a legacy (pre eth/72) peer re-reads it from disk
and re-encodes its full blobs on every GetPooledTransactions request, 
even though the same transaction is typically requested by multiple peers. 
The fast path removed the KZG cost, but the disk read and blob re-encode 
would remain without this. Enhance, a dedicated cache is added for 
this purpose.

Author: cskiraly

core/txpool/blobpool/blobpool.go

@@ -31,6 +31,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
 	"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
 	"github.com/ethereum/go-ethereum/core"
@@ -83,6 +84,11 @@ const (
 	// so pushing it down too aggressively might make resurrections non-functional.
 	maxTxsPerAccount = 16
 
+	// getRLPCacheSize is the byte budget of the cache of encoded
+	// GetPooledTransactions responses. It bounds the memory spent on serving
+	// the same blob transactions to multiple (legacy) peers.
+	getRLPCacheSize = 16 * 1024 * 1024
+
 	// pendingTransactionStore is the subfolder containing the currently queued
 	// blob transactions.
 	pendingTransactionStore = "queue"
@@ -575,9 +581,22 @@ type BlobPool struct {
 	discoverFeed event.Feed // Event feed to send out new tx events on pool discovery (reorg excluded)
 	insertFeed   event.Feed // Event feed to send out new tx events on pool inclusion (reorg included)
 
+	// rlpCache memoizes encoded GetPooledTransactions responses, keyed by
+	// (tx hash, whether full blobs are needed). It is content-addressed: a
+	// pooled tx's encoding is stable, so a cached entry is valid as long as the
+	// tx is still in the pool (checked on lookup before serving a hit).
+	rlpCache *lru.SizeConstrainedCache[rlpCacheKey, []byte]
+
 	lock sync.RWMutex // Mutex protecting the pool during reorg handling
 }
 
+// rlpCacheKey keys the encoded-response cache. full is true for pre eth/72
+// requests (which carry full blobs) and false for eth/72+ (blob payload elided).
+type rlpCacheKey struct {
+	hash common.Hash
+	full bool
+}
+
 // New creates a new blob transaction pool to gather, sort and filter inbound
 // blob transactions from the network.
 func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bool) *BlobPool {
@@ -595,6 +614,7 @@ func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bo
 		spent:          make(map[common.Address]*uint256.Int),
 		gapped:         make(map[common.Address][]*BlobTxForPool),
 		gappedSource:   make(map[common.Hash]common.Address),
+		rlpCache:       lru.NewSizeConstrainedCache[rlpCacheKey, []byte](getRLPCacheSize),
 	}
 }
 
@@ -1753,17 +1773,39 @@ func (p *BlobPool) Get(hash common.Hash) *types.Transaction {
 }
 
 // GetRLP returns an RLP-encoded transaction if it is contained in the pool.
+//
+// The encoded response is memoized: legacy (pre eth/72) peers fetch full blobs
+// via GetPooledTransactions, and the same transaction is typically requested
+// by multiple peers. Without the cache, each request re-reads the tx from disk
+// and re-encodes it, recovering the blobs from the stored cells.
 func (p *BlobPool) GetRLP(hash common.Hash, version uint) []byte {
+	key := rlpCacheKey{hash: hash, full: version < 72}
+	if enc, ok := p.rlpCache.Get(key); ok {
+		// The encoding is content-addressed, but only serve it while the tx is
+		// still pooled so a hit cannot resurrect a dropped transaction.
+		p.lock.RLock()
+		_, pooled := p.lookup.storeidOfTx(hash)
+		p.lock.RUnlock()
+		if pooled {
+			getRLPCacheHitMeter.Mark(1)
+			return enc
+		}
+	}
 	data := p.getRLP(hash)
 	if len(data) == 0 {
 		// Not in this pool, do not log.
 		return nil
 	}
+	// Count misses only for transactions the pool actually holds, so the
+	// hit/miss ratio reflects cache effectiveness rather than unknown-tx
+	// requests.
+	getRLPCacheMissMeter.Mark(1)
 	rlp, err := encodeForNetwork(data, version)
 	if err != nil {
 		log.Error("Failed to encode pooled tx into the network type", "hash", hash, "err", err)
 		return nil
 	}
+	p.rlpCache.Add(key, rlp)
 	return rlp
 }
 

core/txpool/blobpool/blobpool_test.go

@@ -2439,3 +2439,66 @@ func TestGetCells(t *testing.T) {
 		})
 	}
 }
+
+// TestGetRLPCache checks that GetRLP memoizes encoded responses, keys full (pre
+// eth/72) and sparse (eth/72+) encodings separately, and does not serve a cached
+// entry once the transaction has left the pool.
+func TestGetRLPCache(t *testing.T) {
+	storage := t.TempDir()
+	os.MkdirAll(filepath.Join(storage, pendingTransactionStore), 0700)
+	store, _ := billy.Open(billy.Options{Path: filepath.Join(storage, pendingTransactionStore)}, newSlotterEIP7594(params.BlobTxMaxBlobs), nil)
+
+	var (
+		key1, _  = crypto.GenerateKey()
+		addr1    = crypto.PubkeyToAddress(key1.PublicKey)
+		tx1      = makeMultiBlobTx(0, 1, 1000, 100, 1, 0, key1)
+		ptx1, _  = newBlobTxForPool(tx1)
+		blob1, _ = rlp.EncodeToBytes(ptx1)
+	)
+	store.Put(blob1)
+	store.Close()
+
+	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
+	statedb.AddBalance(addr1, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified)
+	statedb.Commit(params.Rules{IsEIP158: true}, 0)
+	chain := &testBlockChain{
+		config:  params.MainnetChainConfig,
+		basefee: uint256.NewInt(params.InitialBaseFee),
+		blobfee: uint256.NewInt(params.BlobTxMinBlobGasprice),
+		statedb: statedb,
+	}
+	pool := New(Config{Datadir: storage}, chain, nil)
+	if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil {
+		t.Fatalf("failed to create blob pool: %v", err)
+	}
+	defer pool.Close()
+
+	// Full (legacy) encoding is memoized and stable across requests.
+	full1 := pool.GetRLP(tx1.Hash(), 71)
+	if len(full1) == 0 {
+		t.Fatalf("expected full encoding, got empty")
+	}
+	full2 := pool.GetRLP(tx1.Hash(), 71)
+	if !bytes.Equal(full1, full2) {
+		t.Fatalf("cached full encoding differs from first request")
+	}
+	// Same backing array proves the repeat response was served from the cache
+	// rather than re-read from disk and re-encoded.
+	if &full1[0] != &full2[0] {
+		t.Fatalf("repeat request was re-encoded instead of served from the cache")
+	}
+	// Sparse (eth/72+) encoding omits blob payloads: smaller and cached separately.
+	sparse := pool.GetRLP(tx1.Hash(), 72)
+	if len(sparse) == 0 || len(sparse) >= len(full1) {
+		t.Fatalf("expected smaller sparse encoding, got %d vs full %d", len(sparse), len(full1))
+	}
+	// Unknown transaction is not served.
+	if got := pool.GetRLP(common.Hash{0xde, 0xad}, 71); got != nil {
+		t.Fatalf("expected nil for unknown tx, got %d bytes", len(got))
+	}
+	// After the tx leaves the pool, a cached entry must not be served.
+	pool.Clear()
+	if got := pool.GetRLP(tx1.Hash(), 71); got != nil {
+		t.Fatalf("expected nil after the tx left the pool, got %d bytes", len(got))
+	}
+}

core/txpool/blobpool/metrics.go

@@ -33,6 +33,10 @@ var (
 	limboDatarealGauge = metrics.NewRegisteredGauge("blobpool/limbo/datareal", nil)
 	limboSlotusedGauge = metrics.NewRegisteredGauge("blobpool/limbo/slotused", nil)
 
+	// The below metrics track the encoded GetPooledTransactions response cache.
+	getRLPCacheHitMeter  = metrics.NewRegisteredMeter("blobpool/getrlp/hit", nil)
+	getRLPCacheMissMeter = metrics.NewRegisteredMeter("blobpool/getrlp/miss", nil)
+
 	// The below metrics track the per-shelf metrics for the primary blob store
 	// and the temporary limbo store.
 	shelfDatausedGaugeName = "blobpool/shelf_%d/dataused"