diff --git a/core/txpool/blobpool/buffer.go b/core/txpool/blobpool/buffer.go index d80e793b821..4b4403b52bd 100644 --- a/core/txpool/blobpool/buffer.go +++ b/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 +} diff --git a/core/txpool/blobpool/buffer_test.go b/core/txpool/blobpool/buffer_test.go index beba17f0a48..11899a909e9 100644 --- a/core/txpool/blobpool/buffer_test.go +++ b/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)) + } +} diff --git a/core/txpool/blobpool/provider_serving_test.go b/core/txpool/blobpool/provider_serving_test.go new file mode 100644 index 00000000000..01e9f135349 --- /dev/null +++ b/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 . + +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) + } + } + } +}