core/txpool/blobpool: extend fully-fetched blob transactions to the full cell set

core/txpool/blobpool/buffer.go

@@ -37,6 +37,8 @@ var (
 	blobBufferCellsFirstCounter = metrics.NewRegisteredCounter("blobpool/buffer/cellsfirst", nil)
 	blobBufferTotalTx           = metrics.NewRegisteredGauge("blobpool/buffer/txcount", nil)
 	blobBufferTotalCells        = metrics.NewRegisteredGauge("blobpool/buffer/cellcount", nil)
+	blobBufferExtendedCounter   = metrics.NewRegisteredCounter("blobpool/buffer/extended", nil)
+	blobBufferExtendFailCounter = metrics.NewRegisteredCounter("blobpool/buffer/extendfail", nil)
 )
 
 const (
@@ -167,8 +169,10 @@ func (b *BlobBuffer) AddCells(hash common.Hash, deliveries map[string]*PeerDeliv
 	blobBufferCellsFirstCounter.Inc(1)
 }
 
-// storeCompleted verifies cells per-peer, sorts them, and schedules them for
-// addition into the pool. The actual addition happens in Flush().
+// storeCompleted verifies cells per-peer, sorts them, completes the extended
+// cell set when the collected cells suffice to reconstruct it (provider
+// extension), and schedules the transaction for addition into the pool. The
+// actual addition happens in Flush().
 func (b *BlobBuffer) storeCompleted(hash common.Hash, tx *types.Transaction, cells *cellEntry) {
 	sidecar := tx.BlobTxSidecar()
 
@@ -182,6 +186,27 @@ func (b *BlobBuffer) storeCompleted(hash common.Hash, tx *types.Transaction, cel
 	blobCount := len(tx.BlobHashes())
 	sorted, custody := sortCells(cells, blobCount)
 
+	// Provider extension: if the collected cells suffice to reconstruct the
+	// blobs but don't cover the full extended set, complete it locally. This
+	// trades a few ms of compute for not downloading the remaining cells, and
+	// lets the node announce -- and serve -- full availability, as the provider
+	// role requires. The reconstructed cells are determined by the already-
+	// verified input, but the shipped proofs of the non-custodied indices have
+	// not been verified yet, so check them before adopting: a transaction whose
+	// own proofs don't match its data is invalid and is discarded.
+	if n := custody.OneCount(); n >= kzg4844.DataPerBlob && n < kzg4844.CellsPerBlob {
+		extended, err := extendCells(sidecar, sorted, custody)
+		if err != nil {
+			log.Warn("Dropping blob tx with unverifiable extension proofs", "hash", hash, "err", err)
+			blobBufferExtendFailCounter.Inc(1)
+			delete(b.cells, hash)
+			delete(b.txs, hash)
+			return
+		}
+		sorted, custody = extended, types.CustodyBitmapAll
+		blobBufferExtendedCounter.Inc(1)
+	}
+
 	cellSidecar := types.BlobTxCellSidecar{
 		Version:     sidecar.Version,
 		Commitments: sidecar.Commitments,
@@ -316,3 +341,35 @@ func sortCells(entry *cellEntry, blobCount int) ([]kzg4844.Cell, types.CustodyBi
 	custody := types.NewCustodyBitmap(indices)
 	return res, custody
 }
+
+// extendCells completes a reconstructable cell set (at least DataPerBlob cells
+// per blob, in ascending custody order, already proof-verified by the caller)
+// to the full extended set, and verifies the newly produced cells against the
+// transaction's shipped proofs. Those proofs could not be checked before: their
+// cells did not exist until reconstruction.
+func extendCells(sidecar *types.BlobTxSidecar, cells []kzg4844.Cell, custody types.CustodyBitmap) ([]kzg4844.Cell, error) {
+	blobCount := len(sidecar.Commitments)
+	if len(sidecar.Proofs) != blobCount*kzg4844.CellProofsPerBlob {
+		return nil, fmt.Errorf("invalid number of cell proofs: %d", len(sidecar.Proofs))
+	}
+	all, err := kzg4844.RecoverCells(cells, custody.Indices())
+	if err != nil {
+		return nil, err
+	}
+	// Verify the not-previously-covered indices in one batch.
+	var (
+		newIndices = types.CustodyBitmapAll.Difference(custody).Indices()
+		vcells     = make([]kzg4844.Cell, 0, blobCount*len(newIndices))
+		vproofs    = make([]kzg4844.Proof, 0, blobCount*len(newIndices))
+	)
+	for b := range blobCount {
+		for _, idx := range newIndices {
+			vcells = append(vcells, all[b*kzg4844.CellsPerBlob+int(idx)])
+			vproofs = append(vproofs, sidecar.Proofs[b*kzg4844.CellProofsPerBlob+int(idx)])
+		}
+	}
+	if err := kzg4844.VerifyCells(vcells, sidecar.Commitments, vproofs, newIndices); err != nil {
+		return nil, err
+	}
+	return all, nil
+}

core/txpool/blobpool/buffer_test.go

@@ -204,3 +204,134 @@ func TestBadCell(t *testing.T) {
 		t.Fatal("buffer should be empty after bad cell drop")
 	}
 }
+
+// TestProviderExtension checks that a transaction completed with the data cells
+// only (a full fetch) is extended to the full cell set before being handed to
+// the pool, with all-ones custody and cells matching the ground truth.
+func TestProviderExtension(t *testing.T) {
+	key, _ := crypto.GenerateKey()
+	blobCount := 2
+
+	var stored []*BlobTxForPool
+	buf := NewBlobBuffer(BlobBufferFunctions{
+		ValidateTx: func(tx *types.Transaction) error { return nil },
+		AddToPool:  func(ptx *BlobTxForPool) error { stored = append(stored, ptx); return nil },
+		DropPeer:   func(peer string) {},
+	})
+
+	tx := makeV1Tx(t, 0, blobCount, 0, key)
+	hash := tx.Hash()
+	buf.AddTx([]*types.Transaction{tx}, "peerA")
+
+	dataIndices := make([]uint64, kzg4844.DataPerBlob)
+	for i := range dataIndices {
+		dataIndices[i] = uint64(i)
+	}
+	delivery := makePeerDelivery(t, 0, blobCount, dataIndices)
+	buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, types.NewCustodyBitmap(dataIndices))
+
+	if buf.HasTx(hash) || buf.HasCells(hash) {
+		t.Fatal("buffer should be empty after completion")
+	}
+	buf.Flush()
+	if len(stored) != 1 {
+		t.Fatalf("expected 1 stored tx, got %d", len(stored))
+	}
+	cs := stored[0].CellSidecar
+	if cs.Custody != types.CustodyBitmapAll {
+		t.Fatalf("custody not extended to all-ones: %d cells", cs.Custody.OneCount())
+	}
+	if len(cs.Cells) != blobCount*kzg4844.CellsPerBlob {
+		t.Fatalf("expected %d cells, got %d", blobCount*kzg4844.CellsPerBlob, len(cs.Cells))
+	}
+	// Compare every cell (data and extension) against the ground truth.
+	for b := 0; b < blobCount; b++ {
+		truth, err := kzg4844.ComputeCells([]kzg4844.Blob{*testBlobs[b]})
+		if err != nil {
+			t.Fatal(err)
+		}
+		for i := range truth {
+			if cs.Cells[b*kzg4844.CellsPerBlob+i] != truth[i] {
+				t.Fatalf("blob %d cell %d does not match ground truth", b, i)
+			}
+		}
+	}
+}
+
+// TestProviderExtensionBadProof checks that a transaction whose shipped
+// extension proofs don't match its (verified) data is discarded rather than
+// stored: the extension cells are correct by construction, so a proof mismatch
+// means the transaction's own sidecar is inconsistent.
+func TestProviderExtensionBadProof(t *testing.T) {
+	key, _ := crypto.GenerateKey()
+	blobCount := 1
+
+	var (
+		stored  []*BlobTxForPool
+		dropped []string
+	)
+	buf := NewBlobBuffer(BlobBufferFunctions{
+		ValidateTx: func(tx *types.Transaction) error { return nil },
+		AddToPool:  func(ptx *BlobTxForPool) error { stored = append(stored, ptx); return nil },
+		DropPeer:   func(peer string) { dropped = append(dropped, peer) },
+	})
+
+	tx := makeV1Tx(t, 0, blobCount, 0, key)
+	// Corrupt one extension proof (an index outside the delivered custody, so
+	// per-peer verification cannot catch it).
+	tx.BlobTxSidecar().Proofs[kzg4844.DataPerBlob][0] ^= 0xff
+	hash := tx.Hash()
+	buf.AddTx([]*types.Transaction{tx}, "peerA")
+
+	dataIndices := make([]uint64, kzg4844.DataPerBlob)
+	for i := range dataIndices {
+		dataIndices[i] = uint64(i)
+	}
+	delivery := makePeerDelivery(t, 0, blobCount, dataIndices)
+	buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, types.NewCustodyBitmap(dataIndices))
+
+	buf.Flush()
+	if len(stored) != 0 {
+		t.Fatalf("tx with bad extension proof must not be stored, got %d", len(stored))
+	}
+	if len(dropped) != 0 {
+		t.Fatalf("cell-delivering peers must not be dropped for a bad tx sidecar, got: %v", dropped)
+	}
+	if buf.HasTx(hash) || buf.HasCells(hash) {
+		t.Fatal("buffer should be empty after drop")
+	}
+}
+
+// TestNoExtensionBelowThreshold checks that a partial (sampler) transaction
+// with fewer than DataPerBlob cells is stored as delivered, unextended.
+func TestNoExtensionBelowThreshold(t *testing.T) {
+	key, _ := crypto.GenerateKey()
+	blobCount := 2
+
+	var stored []*BlobTxForPool
+	buf := NewBlobBuffer(BlobBufferFunctions{
+		ValidateTx: func(tx *types.Transaction) error { return nil },
+		AddToPool:  func(ptx *BlobTxForPool) error { stored = append(stored, ptx); return nil },
+		DropPeer:   func(peer string) {},
+	})
+
+	tx := makeV1Tx(t, 0, blobCount, 0, key)
+	hash := tx.Hash()
+	buf.AddTx([]*types.Transaction{tx}, "peerA")
+
+	indices := []uint64{3, 17, 64, 100} // sampler custody, incl. extension columns
+	delivery := makePeerDelivery(t, 0, blobCount, indices)
+	buf.AddCells(hash, map[string]*PeerDelivery{"peerB": delivery}, types.NewCustodyBitmap(indices))
+
+	buf.Flush()
+	if len(stored) != 1 {
+		t.Fatalf("expected 1 stored tx, got %d", len(stored))
+	}
+	cs := stored[0].CellSidecar
+	if cs.Custody != types.NewCustodyBitmap(indices) {
+		t.Fatalf("sampler custody must be stored as delivered")
+	}
+	if len(cs.Cells) != blobCount*len(indices) {
+		t.Fatalf("expected %d cells, got %d", blobCount*len(indices), len(cs.Cells))
+	}
+}

core/txpool/blobpool/provider_serving_test.go

@@ -0,0 +1,140 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package blobpool
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/tracing"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/holiman/billy"
+	"github.com/holiman/uint256"
+)
+
+// TestProviderExtensionServing checks, end to end, that a transaction acquired
+// as a full fetch (only the data cells delivered) ends up fully servable and
+// fully advertised: the buffer completes it, the pool stores it, and
+//
+//   - GetCustody reports all-ones -- the mask the tx announcement carries, so
+//     peers may request any column, and
+//   - GetBlobCells serves extension columns (>= DataPerBlob) byte-correctly --
+//     the exact call behind both the eth GetCells handler (p2p serving) and the
+//     engine_getBlobsV4 cache miss path (CL serving).
+//
+// This test intentionally uses only APIs that predate the provider-extension
+// change, so it can run against older code to demonstrate the gap: there, the
+// stored/advertised custody is just the data cells and extension columns are
+// returned as nil.
+func TestProviderExtensionServing(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)
+	store.Close()
+
+	var (
+		key, _    = crypto.GenerateKey()
+		addr      = crypto.PubkeyToAddress(key.PublicKey)
+		blobCount = 2
+		// Post-eth/72 shape: blob payload elided, commitments and cell proofs kept.
+		tx   = removeBlobs(makeMultiBlobTx(0, 10, 2*params.InitialBaseFee, 100, blobCount, 0, key))
+		hash = tx.Hash()
+	)
+	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
+	statedb.AddBalance(addr, uint256.NewInt(1_000_000_000_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()
+
+	// Wire the buffer to the real pool, as the eth handler does, and run the
+	// full-fetch ingest flow: tx body first, then a data-cells-only delivery.
+	buf := NewBlobBuffer(BlobBufferFunctions{
+		ValidateTx: pool.ValidateTxBasics,
+		AddToPool:  pool.AddPooledTx,
+		DropPeer:   func(peer string) {},
+	})
+	if err := buf.AddTx([]*types.Transaction{tx}, "peerA")[0]; err != nil {
+		t.Fatalf("AddTx: %v", err)
+	}
+	dataIndices := make([]uint64, kzg4844.DataPerBlob)
+	for i := range dataIndices {
+		dataIndices[i] = uint64(i)
+	}
+	buf.AddCells(hash, map[string]*PeerDelivery{"peerB": makePeerDelivery(t, 0, blobCount, dataIndices)}, types.NewCustodyBitmap(dataIndices))
+
+	hashes, errs := buf.Flush()
+	if len(hashes) != 1 || errs[0] != nil {
+		t.Fatalf("expected 1 pooled tx, got %d (err %v)", len(hashes), errs)
+	}
+
+	// Advertising: the announcement mask is built from GetCustody, so all-ones
+	// here means peers are invited to request any column.
+	if custody := pool.GetCustody(hash); custody == nil || *custody != types.CustodyBitmapAll {
+		have := -1
+		if custody != nil {
+			have = custody.OneCount()
+		}
+		t.Errorf("advertised custody not all-ones: have %d cells, want %d", have, kzg4844.CellsPerBlob)
+	}
+
+	// Serving: request extension columns the node never downloaded.
+	extIndices := make([]uint64, 32)
+	for i := range extIndices {
+		extIndices[i] = uint64(kzg4844.DataPerBlob + i)
+	}
+	vhashes := pool.GetBlobHashes(hash)
+	if len(vhashes) != blobCount {
+		t.Fatalf("expected %d versioned hashes, got %d", blobCount, len(vhashes))
+	}
+	cells, proofs, err := pool.GetBlobCells(vhashes, types.NewCustodyBitmap(extIndices))
+	if err != nil {
+		t.Fatalf("GetBlobCells: %v", err)
+	}
+	for b := 0; b < blobCount; b++ {
+		truth, err := kzg4844.ComputeCells([]kzg4844.Blob{*testBlobs[b]})
+		if err != nil {
+			t.Fatal(err)
+		}
+		for k, idx := range extIndices {
+			if cells[b] == nil || cells[b][k] == nil {
+				t.Errorf("blob %d: extension cell %d not served", b, idx)
+				continue
+			}
+			if *cells[b][k] != truth[idx] {
+				t.Errorf("blob %d: extension cell %d does not match ground truth", b, idx)
+			}
+			if proofs[b] == nil || proofs[b][k] == nil {
+				t.Errorf("blob %d: extension proof %d not served", b, idx)
+			}
+		}
+	}
+}