core, eth: answer GetBlobsV4 from cache

core/txpool/blobpool/blobpool.go

@@ -38,7 +38,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
-	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
@@ -291,7 +290,8 @@ func encodeForNetwork(storedRLP []byte, version uint) ([]byte, error) {
 
 	// 5. Build the [blobs] field for the wire format.
 	var blobsField []byte
-	if version >= eth.ETH72 {
+	// todo - Didn't use eth.ETH72 due to circular import error in test
+	if version >= 72 {
 		// eth/72 omits the blob payload; peers fetch cells separately via GetCells.
 		blobsField = []byte{0xc0} // RLP-encoded empty list
 	} else {
@@ -2241,6 +2241,8 @@ func (p *BlobPool) drop() {
 //
 // The transactions can also be pre-filtered by the dynamic fee components to
 // reduce allocations and load on downstream subsystems.
+// The transactions without enough cells to recover thier blobs would be skipped
+// in the response.
 func (p *BlobPool) Pending(filter txpool.PendingFilter) (map[common.Address][]*txpool.LazyTransaction, int) {
 	// If only plain transactions are requested, this pool is unsuitable as it
 	// contains none, don't even bother.
@@ -2295,7 +2297,7 @@ func (p *BlobPool) Pending(filter txpool.PendingFilter) (map[common.Address][]*t
 				}
 			}
 			// Skip transactions without enough cells to recover blobs
-			if tx.custody != nil && tx.custody.OneCount() < kzg4844.DataPerBlob {
+			if !filter.PartialCells && tx.custody != nil && tx.custody.OneCount() < kzg4844.DataPerBlob {
 				break // not enough cells to build a full payload, discard rest of txs from the account
 			}
 			// Transaction was accepted according to the filter, append to the pending list

core/txpool/blobpool/cache.go

@@ -48,10 +48,12 @@ var (
 	// (blobs.filled - cache.hit).
 	cacheHitMeter     = metrics.NewRegisteredMeter("blobpool/cache/hit", nil)
 	cacheMissMeter    = metrics.NewRegisteredMeter("blobpool/cache/miss", nil)
-	cacheBlobsGauge = metrics.NewRegisteredGauge("blobpool/cache/blobs", nil)
+	cacheEntriesGauge = metrics.NewRegisteredGauge("blobpool/cache/entries", nil)
 )
 
 type cachedBlob struct {
+	cell       []kzg4844.Cell
+	custody    types.CustodyBitmap
 	blob       *kzg4844.Blob
 	commitment kzg4844.Commitment
 	proofs     []kzg4844.Proof
@@ -74,11 +76,16 @@ type Cache struct {
 	mu      sync.Mutex
 	entries map[common.Hash]*cachedBlob
 
+	// needCell is owned by the loop goroutine; it is only read and written
+	// there, so it needs no synchronization. It is flipped on via enableCellCh.
+	needCell bool
+
 	// channels into loop
 	quit         chan struct{}
 	topkRequest  chan struct{}
 	topkTimer    mclock.Timer
 	hasBlobsCh   chan []common.Hash // list of tx hashes that should be pinned
+	enableCellCh chan struct{}      // signals the loop to switch to cell mode
 
 	step func() // test hook fired after each loop iteration
 
@@ -103,6 +110,7 @@ func newCache(p *BlobPool, clock mclock.Clock, step func()) *Cache {
 		step:         step,
 		quit:         make(chan struct{}),
 		topkRequest:  make(chan struct{}, 1),
+		enableCellCh: make(chan struct{}, 1),
 	}
 
 	c.wg.Add(1)
@@ -197,13 +205,10 @@ func (c *Cache) GetBlobs(ctx context.Context, vhashes []common.Hash, version byt
 	c.mu.Lock()
 	for vhash, idxs := range indices {
 		n := len(idxs)
-
-		cached := c.entries[vhash]
-		if cached == nil || cached.version != version {
+		cached, ok := c.entries[vhash]
+		if !ok || cached.version != version || cached.blob == nil {
 			cacheMiss += n
-			if cached == nil {
 			misses = append(misses, vhash)
-			}
 			continue
 		}
 		cacheHits += n
@@ -241,6 +246,78 @@ func (c *Cache) GetBlobs(ctx context.Context, vhashes []common.Hash, version byt
 	return blobs, commitments, proofs, nil
 }
 
+func (c *Cache) GetCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]*kzg4844.Cell, [][]*kzg4844.Proof, error) {
+	var (
+		cells   = make([][]*kzg4844.Cell, len(vhashes))
+		proofs  = make([][]*kzg4844.Proof, len(vhashes))
+		indices = make(map[common.Hash][]int)
+		misses  []common.Hash
+	)
+	for i, h := range vhashes {
+		indices[h] = append(indices[h], i)
+	}
+	requested := mask.Indices()
+
+	c.mu.Lock()
+	for vhash, idxs := range indices {
+		cached, ok := c.entries[vhash]
+		if !ok {
+			misses = append(misses, vhash)
+			continue
+		}
+		stored := cached.custody.Indices()
+		blobCells := make([]*kzg4844.Cell, len(requested))
+		blobProofs := make([]*kzg4844.Proof, len(requested))
+		for i, cellIdx := range requested {
+			pos := -1
+			for k, s := range stored {
+				if s == cellIdx {
+					pos = k
+					break
+				}
+			}
+			if pos >= 0 && pos < len(cached.cell) {
+				cell := cached.cell[pos]
+				blobCells[i] = &cell
+				if int(cellIdx) < len(cached.proofs) {
+					pf := cached.proofs[cellIdx]
+					blobProofs[i] = &pf
+				}
+			}
+		}
+		for _, idx := range idxs {
+			cells[idx] = blobCells
+			proofs[idx] = blobProofs
+		}
+	}
+	c.mu.Unlock()
+
+	if len(misses) > 0 {
+		mc, mp, err := c.blobpool.GetBlobCells(misses, mask)
+		if err != nil {
+			return nil, nil, err
+		}
+		for j, vhash := range misses {
+			for _, idx := range indices[vhash] {
+				cells[idx] = mc[j]
+				proofs[idx] = mp[j]
+			}
+		}
+	}
+	return cells, proofs, nil
+}
+
+// EnableCell allows the cache to store only cells without recovering
+// blobs. This means we can also cache cells that lack enough blobs to
+// recover. It signals the loop to switch to cell mode and re-select
+// transactions from this wider pool.
+func (c *Cache) EnableCell() {
+	select {
+	case c.enableCellCh <- struct{}{}:
+	case <-c.quit:
+	}
+}
+
 func (c *Cache) loop() {
 	defer c.wg.Done()
 
@@ -258,6 +335,16 @@ func (c *Cache) loop() {
 			c.update(want)
 			c.triggerTopKAfter(topKTimeout)
 
+		case <-c.enableCellCh:
+			// CL supports cell-based blob retrieval; switch to cell mode and
+			// re-select immediately over the now-wider pool. needCell is only
+			// touched here in the loop, so a fresh selection and update observe
+			// a consistent value.
+			if !c.needCell {
+				c.needCell = true
+				c.triggerTopK()
+			}
+
 		case <-c.quit:
 			c.cancelUpdate()
 			if c.topkTimer != nil {
@@ -285,6 +372,8 @@ func (c *Cache) cancelUpdate() {
 // are no longer wanted and loads the missing ones from the blobpool in the
 // background.
 func (c *Cache) update(want []common.Hash) {
+	cellMode := c.needCell
+
 	wantSet := make(map[common.Hash]struct{}, len(want))
 	for _, vh := range want {
 		wantSet[vh] = struct{}{}
@@ -298,7 +387,13 @@ func (c *Cache) update(want []common.Hash) {
 	c.mu.Lock()
 	var missing []common.Hash
 	for vh := range wantSet {
-		if _, ok := c.entries[vh]; !ok {
+		e, ok := c.entries[vh]
+		if ok && ((cellMode && e.cell == nil) || (!cellMode && e.blob == nil)) {
+			delete(c.entries, vh)
+			cacheEntriesGauge.Dec(1)
+			ok = false
+		}
+		if !ok {
 			missing = append(missing, vh)
 		}
 	}
@@ -307,7 +402,7 @@ func (c *Cache) update(want []common.Hash) {
 			continue
 		}
 		delete(c.entries, vh)
-		cacheBlobsGauge.Dec(1)
+		cacheEntriesGauge.Dec(1)
 	}
 	c.mu.Unlock()
 
@@ -330,18 +425,69 @@ func (c *Cache) update(want []common.Hash) {
 			if ptx == nil {
 				continue
 			}
-			sidecar, err := ptx.sidecar()
-			if err != nil || sidecar == nil {
+			if cellMode {
+				c.loadCells(ptx, wantSet)
+			} else {
+				c.loadBlobs(ptx, wantSet)
+			}
+		}
+	}()
+}
+
+// loadCells loads the cells whose vhash is in wantSet from ptx.
+func (c *Cache) loadCells(ptx *BlobTxForPool, wantSet map[common.Hash]struct{}) {
+	cs := ptx.CellSidecar
+	cellsPerBlob := cs.Custody.OneCount()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for i, v := range ptx.Tx.BlobHashes() {
+		if _, ok := wantSet[v]; !ok {
 			continue
 		}
+		if _, exists := c.entries[v]; exists {
+			continue
+		}
+		cellStart := i * cellsPerBlob
+		if cellStart+cellsPerBlob > len(cs.Cells) {
+			continue
+		}
+		blobCells := make([]kzg4844.Cell, cellsPerBlob)
+		copy(blobCells, cs.Cells[cellStart:cellStart+cellsPerBlob])
 
+		var pf []kzg4844.Proof
+		if ps := i * kzg4844.CellProofsPerBlob; ps+kzg4844.CellProofsPerBlob <= len(cs.Proofs) {
+			pf = make([]kzg4844.Proof, kzg4844.CellProofsPerBlob)
+			copy(pf, cs.Proofs[ps:ps+kzg4844.CellProofsPerBlob])
+		}
+		c.entries[v] = &cachedBlob{
+			cell:       blobCells,
+			custody:    cs.Custody,
+			commitment: cs.Commitments[i],
+			proofs:     pf,
+			version:    cs.Version,
+		}
+		cacheEntriesGauge.Inc(1)
+	}
+}
+
+// loadBlobs loads the blobs whose vhash is in wantSet from ptx.
+func (c *Cache) loadBlobs(ptx *BlobTxForPool, wantSet map[common.Hash]struct{}) {
+	if ptx.CellSidecar.Custody.OneCount() < kzg4844.DataPerBlob {
+		return
+	}
+	// blobs will be computed inside of sidecar()
+	sidecar, err := ptx.sidecar()
+	if err != nil || sidecar == nil {
+		return
+	}
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	for i, v := range sidecar.BlobHashes() {
 		if _, ok := wantSet[v]; !ok {
 			continue
 		}
 		if _, exists := c.entries[v]; exists {
-					continue // recompute only new entries
+			continue
 		}
 		var pf []kzg4844.Proof
 		switch sidecar.Version {
@@ -361,11 +507,8 @@ func (c *Cache) update(want []common.Hash) {
 			proofs:     pf,
 			version:    sidecar.Version,
 		}
-				cacheBlobsGauge.Inc(1)
+		cacheEntriesGauge.Inc(1)
 	}
-			c.mu.Unlock()
-		}
-	}()
 }
 
 // selectTopTxs returns the vhashes of the top K most profitable pending blob
@@ -382,6 +525,7 @@ func (c *Cache) selectTopTxs() []common.Hash {
 	filter := txpool.PendingFilter{
 		BlobTxs:      true,
 		BaseFee:      uint256.MustFromBig(baseFee),
+		PartialCells: c.needCell,
 	}
 	if head.ExcessBlobGas != nil {
 		filter.BlobFee = uint256.MustFromBig(eip4844.CalcBlobFee(config, head))

core/txpool/subpool.go

@@ -81,6 +81,7 @@ type PendingFilter struct {
 	// When BlobTxs true, return only blob transactions (block blob-space filling)
 	// when false, return only non-blob txs (peer-join announces, block space filling)
 	BlobTxs      bool
+	PartialCells bool
 	BlobVersion  byte // Blob tx version to include. 0 means pre-Osaka, 1 means Osaka and later
 }
 

eth/catalyst/api.go

@@ -724,7 +724,7 @@ func (api *ConsensusAPI) GetBlobsV4(hashes []common.Hash, indicesBitarray types.
 	if len(hashes) > 128 {
 		return nil, engine.TooLargeRequest.With(fmt.Errorf("requested blob count too large: %v", len(hashes)))
 	}
-	cells, proofs, err := api.eth.BlobTxPool().GetBlobCells(hashes, indicesBitarray)
+	cells, proofs, err := api.eth.BlobCache().GetCells(hashes, indicesBitarray)
 	if err != nil {
 		return nil, engine.InvalidParams.With(err)
 	}
@@ -1195,17 +1195,30 @@ func (api *ConsensusAPI) checkFork(timestamp uint64, forks ...forks.Fork) bool {
 }
 
 // ExchangeCapabilities returns the current methods provided by this node.
-func (api *ConsensusAPI) ExchangeCapabilities([]string) []string {
+func (api *ConsensusAPI) ExchangeCapabilities(caps []string) []string {
 	valueT := reflect.TypeOf(api)
-	caps := make([]string, 0, valueT.NumMethod())
+
+	for _, cap := range caps {
+		if cap == "engine_getBlobsV4" {
+			// If the CL supports getBlobsV4, we call EnableCell() on the
+			// blob cache to skip the blob recovery process. This is a
+			// one-directional toggle, which assumes that once the CL
+			// supports getBlobsV4, it will not fall back to getBlobsV3
+			// again.
+			api.eth.BlobCache().EnableCell()
+			break
+		}
+	}
+
+	ourCaps := make([]string, 0, valueT.NumMethod())
 	for i := 0; i < valueT.NumMethod(); i++ {
 		name := []rune(valueT.Method(i).Name)
 		if string(name) == "ExchangeCapabilities" {
 			continue
 		}
-		caps = append(caps, "engine_"+string(unicode.ToLower(name[0]))+string(name[1:]))
+		ourCaps = append(ourCaps, "engine_"+string(unicode.ToLower(name[0]))+string(name[1:]))
 	}
-	return caps
+	return ourCaps
 }
 
 // GetClientVersionV1 exchanges client version data of this node.