core/txpool/blobpool, eth/catalyst: place null for missing blob (#32536)

This pull request fixes a regression, introduced in #32190

Specifically, in GetBlobsV1 engine API, if any blob is missing, the null
should be placed in
response, unfortunately a behavioral change was introduced in #32190,
returning an error
instead.

What's more, a more comprehensive test suite is added to cover both
`GetBlobsV1` and
`GetBlobsV2` endpoints.

core/txpool/blobpool/blobpool.go

@@ -1298,6 +1298,13 @@ func (p *BlobPool) GetMetadata(hash common.Hash) *txpool.TxMetadata {
 }
 
 // GetBlobs returns a number of blobs and proofs for the given versioned hashes.
+// Blobpool must place responses in the order given in the request, using null
+// for any missing blobs.
+//
+// For instance, if the request is [A_versioned_hash, B_versioned_hash,
+// C_versioned_hash] and blobpool has data for blobs A and C, but doesn't have
+// data for B, the response MUST be [A, null, C].
+//
 // This is a utility method for the engine API, enabling consensus clients to
 // retrieve blobs from the pools directly instead of the network.
 func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blob, []kzg4844.Commitment, [][]kzg4844.Proof, error) {
@@ -1317,12 +1324,13 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blo
 		if _, ok := filled[vhash]; ok {
 			continue
 		}
-		// Retrieve the corresponding blob tx with the vhash
+		// Retrieve the corresponding blob tx with the vhash, skip blob resolution
+		// if it's not found locally and place the null instead.
 		p.lock.RLock()
 		txID, exists := p.lookup.storeidOfBlob(vhash)
 		p.lock.RUnlock()
 		if !exists {
-			return nil, nil, nil, fmt.Errorf("blob with vhash %x is not found", vhash)
+			continue
 		}
 		data, err := p.store.Get(txID)
 		if err != nil {

core/txpool/blobpool/blobpool_test.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"math"
 	"math/big"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -41,6 +42,7 @@ import (
 	"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/internal/testrand"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/holiman/billy"
@@ -1814,10 +1816,10 @@ func TestGetBlobs(t *testing.T) {
 	}
 
 	cases := []struct {
-		start   int
+		start      int
-		limit   int
+		limit      int
-		version byte
+		fillRandom bool
-		expErr  bool
+		version    byte
 	}{
 		{
 			start: 0, limit: 6,
@@ -1827,6 +1829,14 @@ func TestGetBlobs(t *testing.T) {
 			start: 0, limit: 6,
 			version: types.BlobSidecarVersion1,
 		},
+		{
+			start: 0, limit: 6, fillRandom: true,
+			version: types.BlobSidecarVersion0,
+		},
+		{
+			start: 0, limit: 6, fillRandom: true,
+			version: types.BlobSidecarVersion1,
+		},
 		{
 			start: 3, limit: 9,
 			version: types.BlobSidecarVersion0,
@@ -1835,6 +1845,14 @@ func TestGetBlobs(t *testing.T) {
 			start: 3, limit: 9,
 			version: types.BlobSidecarVersion1,
 		},
+		{
+			start: 3, limit: 9, fillRandom: true,
+			version: types.BlobSidecarVersion0,
+		},
+		{
+			start: 3, limit: 9, fillRandom: true,
+			version: types.BlobSidecarVersion1,
+		},
 		{
 			start: 3, limit: 15,
 			version: types.BlobSidecarVersion0,
@@ -1843,6 +1861,14 @@ func TestGetBlobs(t *testing.T) {
 			start: 3, limit: 15,
 			version: types.BlobSidecarVersion1,
 		},
+		{
+			start: 3, limit: 15, fillRandom: true,
+			version: types.BlobSidecarVersion0,
+		},
+		{
+			start: 3, limit: 15, fillRandom: true,
+			version: types.BlobSidecarVersion1,
+		},
 		{
 			start: 0, limit: 18,
 			version: types.BlobSidecarVersion0,
@@ -1852,58 +1878,79 @@ func TestGetBlobs(t *testing.T) {
 			version: types.BlobSidecarVersion1,
 		},
 		{
-			start: 18, limit: 20,
+			start: 0, limit: 18, fillRandom: true,
 			version: types.BlobSidecarVersion0,
-			expErr:  true,
+		},
+		{
+			start: 0, limit: 18, fillRandom: true,
+			version: types.BlobSidecarVersion1,
 		},
 	}
 	for i, c := range cases {
-		var vhashes []common.Hash
+		var (
+			vhashes []common.Hash
+			filled  = make(map[int]struct{})
+		)
+		if c.fillRandom {
+			filled[len(vhashes)] = struct{}{}
+			vhashes = append(vhashes, testrand.Hash())
+		}
 		for j := c.start; j < c.limit; j++ {
 			vhashes = append(vhashes, testBlobVHashes[j])
+			if c.fillRandom && rand.Intn(2) == 0 {
+				filled[len(vhashes)] = struct{}{}
+				vhashes = append(vhashes, testrand.Hash())
+			}
+		}
+		if c.fillRandom {
+			filled[len(vhashes)] = struct{}{}
+			vhashes = append(vhashes, testrand.Hash())
 		}
 		blobs, _, proofs, err := pool.GetBlobs(vhashes, c.version)
+		if err != nil {
+			t.Errorf("Unexpected error for case %d, %v", i, err)
+		}
 
-		if c.expErr {
+		// Cross validate what we received vs what we wanted
-			if err == nil {
+		length := c.limit - c.start
-				t.Errorf("Unexpected return, want error for case %d", i)
+		wantLen := length + len(filled)
-			}
+		if len(blobs) != wantLen || len(proofs) != wantLen {
-		} else {
+			t.Errorf("retrieved blobs/proofs size mismatch: have %d/%d, want %d", len(blobs), len(proofs), wantLen)
-			if err != nil {
+			continue
-				t.Errorf("Unexpected error for case %d, %v", i, err)
+		}
-			}
+
-			// Cross validate what we received vs what we wanted
+		var unknown int
-			length := c.limit - c.start
+		for j := 0; j < len(blobs); j++ {
-			if len(blobs) != length || len(proofs) != length {
+			if _, exist := filled[j]; exist {
-				t.Errorf("retrieved blobs/proofs size mismatch: have %d/%d, want %d", len(blobs), len(proofs), length)
+				if blobs[j] != nil || proofs[j] != nil {
+					t.Errorf("Unexpected blob and proof, item %d", j)
+				}
+				unknown++
 				continue
 			}
-			for j := 0; j < len(blobs); j++ {
+			// If an item is missing, but shouldn't, error
-				// If an item is missing, but shouldn't, error
+			if blobs[j] == nil || proofs[j] == nil {
-				if blobs[j] == nil || proofs[j] == nil {
+				t.Errorf("tracked blob retrieval failed: item %d, hash %x", j, vhashes[j])
-					t.Errorf("tracked blob retrieval failed: item %d, hash %x", j, vhashes[j])
+				continue
-					continue
+			}
+			// Item retrieved, make sure the blob matches the expectation
+			if *blobs[j] != *testBlobs[c.start+j-unknown] {
+				t.Errorf("retrieved blob mismatch: item %d, hash %x", j, vhashes[j])
+				continue
+			}
+			// Item retrieved, make sure the proof matches the expectation
+			if c.version == types.BlobSidecarVersion0 {
+				if proofs[j][0] != testBlobProofs[c.start+j-unknown] {
+					t.Errorf("retrieved proof mismatch: item %d, hash %x", j, vhashes[j])
 				}
-				// Item retrieved, make sure the blob matches the expectation
+			} else {
-				if *blobs[j] != *testBlobs[c.start+j] {
+				want, _ := kzg4844.ComputeCellProofs(blobs[j])
-					t.Errorf("retrieved blob mismatch: item %d, hash %x", j, vhashes[j])
+				if !reflect.DeepEqual(want, proofs[j]) {
-					continue
+					t.Errorf("retrieved proof mismatch: item %d, hash %x", j, vhashes[j])
-				}
-				// Item retrieved, make sure the proof matches the expectation
-				if c.version == types.BlobSidecarVersion0 {
-					if proofs[j][0] != testBlobProofs[c.start+j] {
-						t.Errorf("retrieved proof mismatch: item %d, hash %x", j, vhashes[j])
-					}
-				} else {
-					want, _ := kzg4844.ComputeCellProofs(blobs[j])
-					if !reflect.DeepEqual(want, proofs[j]) {
-						t.Errorf("retrieved proof mismatch: item %d, hash %x", j, vhashes[j])
-					}
 				}
 			}
 		}
 	}
-
 	pool.Close()
 }
 

eth/catalyst/api.go

@@ -458,6 +458,26 @@ func (api *ConsensusAPI) getPayload(payloadID engine.PayloadID, full bool) (*eng
 }
 
 // GetBlobsV1 returns a blob from the transaction pool.
+//
+// Specification:
+//
+// Given an array of blob versioned hashes client software MUST respond with an
+// array of BlobAndProofV1 objects with matching versioned hashes, respecting the
+// order of versioned hashes in the input array.
+//
+// Client software MUST place responses in the order given in the request, using
+// null for any missing blobs. For instance:
+//
+// if the request is [A_versioned_hash, B_versioned_hash, C_versioned_hash] and
+// client software has data for blobs A and C, but doesn't have data for B, the
+// response MUST be [A, null, C].
+//
+// Client software MUST support request sizes of at least 128 blob versioned hashes.
+// The client MUST return -38004: Too large request error if the number of requested
+// blobs is too large.
+//
+// Client software MAY return an array of all null entries if syncing or otherwise
+// unable to serve blob pool data.
 func (api *ConsensusAPI) GetBlobsV1(hashes []common.Hash) ([]*engine.BlobAndProofV1, error) {
 	if len(hashes) > 128 {
 		return nil, engine.TooLargeRequest.With(fmt.Errorf("requested blob count too large: %v", len(hashes)))
@@ -468,6 +488,10 @@ func (api *ConsensusAPI) GetBlobsV1(hashes []common.Hash) ([]*engine.BlobAndProo
 	}
 	res := make([]*engine.BlobAndProofV1, len(hashes))
 	for i := 0; i < len(blobs); i++ {
+		// Skip the non-existing blob
+		if blobs[i] == nil {
+			continue
+		}
 		res[i] = &engine.BlobAndProofV1{
 			Blob:  blobs[i][:],
 			Proof: proofs[i][0][:],
@@ -477,6 +501,33 @@ func (api *ConsensusAPI) GetBlobsV1(hashes []common.Hash) ([]*engine.BlobAndProo
 }
 
 // GetBlobsV2 returns a blob from the transaction pool.
+//
+// Specification:
+// Refer to the specification for engine_getBlobsV1 with changes of the following:
+//
+// Given an array of blob versioned hashes client software MUST respond with an
+// array of BlobAndProofV2 objects with matching versioned hashes, respecting
+// the order of versioned hashes in the input array.
+//
+// Client software MUST return null in case of any missing or older version blobs.
+// For instance,
+//
+//   - if the request is [A_versioned_hash, B_versioned_hash, C_versioned_hash] and
+//     client software has data for blobs A and C, but doesn't have data for B, the
+//     response MUST be null.
+//
+//   - if the request is [A_versioned_hash_for_blob_with_blob_proof], the response
+//     MUST be null as well.
+//
+//     Note, geth internally make the conversion from old version to new one, so the
+//     data will be returned normally.
+//
+// Client software MUST support request sizes of at least 128 blob versioned
+// hashes. The client MUST return -38004: Too large request error if the number
+// of requested blobs is too large.
+//
+// Client software MUST return null if syncing or otherwise unable to serve
+// blob pool data.
 func (api *ConsensusAPI) GetBlobsV2(hashes []common.Hash) ([]*engine.BlobAndProofV2, error) {
 	if len(hashes) > 128 {
 		return nil, engine.TooLargeRequest.With(fmt.Errorf("requested blob count too large: %v", len(hashes)))
@@ -498,6 +549,12 @@ func (api *ConsensusAPI) GetBlobsV2(hashes []common.Hash) ([]*engine.BlobAndProo
 	}
 	res := make([]*engine.BlobAndProofV2, len(hashes))
 	for i := 0; i < len(blobs); i++ {
+		// the blob is missing, return null as response. It should
+		// be caught by `AvailableBlobs` though, perhaps data race
+		// occurs between two calls.
+		if blobs[i] == nil {
+			return nil, nil
+		}
 		var cellProofs []hexutil.Bytes
 		for _, proof := range proofs[i] {
 			cellProofs = append(cellProofs, proof[:])

eth/catalyst/api_test.go

@@ -19,7 +19,9 @@ package catalyst
 import (
 	"bytes"
 	"context"
+	"crypto/ecdsa"
 	crand "crypto/rand"
+	"crypto/sha256"
 	"errors"
 	"fmt"
 	"math/big"
@@ -40,6 +42,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
+	"github.com/ethereum/go-ethereum/internal/testrand"
 	"github.com/ethereum/go-ethereum/internal/version"
 	"github.com/ethereum/go-ethereum/miner"
 	"github.com/ethereum/go-ethereum/node"
@@ -47,6 +50,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/trie"
+	"github.com/holiman/uint256"
 )
 
 var (
@@ -112,7 +116,7 @@ func TestEth2AssembleBlock(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks)
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 	signer := types.NewEIP155Signer(ethservice.BlockChain().Config().ChainID)
 	tx, err := types.SignTx(types.NewTransaction(uint64(10), blocks[9].Coinbase(), big.NewInt(1000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, testKey)
 	if err != nil {
@@ -151,7 +155,7 @@ func TestEth2AssembleBlockWithAnotherBlocksTxs(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks[:9])
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// Put the 10th block's tx in the pool and produce a new block
 	txs := blocks[9].Transactions()
@@ -173,7 +177,7 @@ func TestEth2PrepareAndGetPayload(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks[:9])
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// Put the 10th block's tx in the pool and produce a new block
 	txs := blocks[9].Transactions()
@@ -238,8 +242,9 @@ func TestInvalidPayloadTimestamp(t *testing.T) {
 	genesis, preMergeBlocks := generateMergeChain(10, false)
 	n, ethservice := startEthService(t, genesis, preMergeBlocks)
 	defer n.Close()
+
 	var (
-		api    = NewConsensusAPI(ethservice)
+		api    = newConsensusAPIWithoutHeartbeat(ethservice)
 		parent = ethservice.BlockChain().CurrentBlock()
 	)
 	tests := []struct {
@@ -281,7 +286,7 @@ func TestEth2NewBlock(t *testing.T) {
 	defer n.Close()
 
 	var (
-		api    = NewConsensusAPI(ethservice)
+		api    = newConsensusAPIWithoutHeartbeat(ethservice)
 		parent = preMergeBlocks[len(preMergeBlocks)-1]
 
 		// This EVM code generates a log when the contract is created.
@@ -434,8 +439,14 @@ func startEthService(t *testing.T, genesis *core.Genesis, blocks []*types.Block)
 		t.Fatal("can't create node:", err)
 	}
 
-	mcfg := miner.DefaultConfig
+	ethcfg := &ethconfig.Config{
-	ethcfg := &ethconfig.Config{Genesis: genesis, SyncMode: ethconfig.FullSync, TrieTimeout: time.Minute, TrieDirtyCache: 256, TrieCleanCache: 256, Miner: mcfg}
+		Genesis:        genesis,
+		SyncMode:       ethconfig.FullSync,
+		TrieTimeout:    time.Minute,
+		TrieDirtyCache: 256,
+		TrieCleanCache: 256,
+		Miner:          miner.DefaultConfig,
+	}
 	ethservice, err := eth.New(n, ethcfg)
 	if err != nil {
 		t.Fatal("can't create eth service:", err)
@@ -459,6 +470,7 @@ func TestFullAPI(t *testing.T) {
 	genesis, preMergeBlocks := generateMergeChain(10, false)
 	n, ethservice := startEthService(t, genesis, preMergeBlocks)
 	defer n.Close()
+
 	var (
 		parent = ethservice.BlockChain().CurrentBlock()
 		// This EVM code generates a log when the contract is created.
@@ -476,7 +488,7 @@ func TestFullAPI(t *testing.T) {
 }
 
 func setupBlocks(t *testing.T, ethservice *eth.Ethereum, n int, parent *types.Header, callback func(parent *types.Header), withdrawals [][]*types.Withdrawal, beaconRoots []common.Hash) []*types.Header {
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 	var blocks []*types.Header
 	for i := 0; i < n; i++ {
 		callback(parent)
@@ -524,7 +536,7 @@ func TestExchangeTransitionConfig(t *testing.T) {
 	defer n.Close()
 
 	// invalid ttd
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 	config := engine.TransitionConfigurationV1{
 		TerminalTotalDifficulty: (*hexutil.Big)(big.NewInt(0)),
 		TerminalBlockHash:       common.Hash{},
@@ -585,7 +597,7 @@ func TestNewPayloadOnInvalidChain(t *testing.T) {
 	defer n.Close()
 
 	var (
-		api    = NewConsensusAPI(ethservice)
+		api    = newConsensusAPIWithoutHeartbeat(ethservice)
 		parent = ethservice.BlockChain().CurrentBlock()
 		signer = types.LatestSigner(ethservice.BlockChain().Config())
 		// This EVM code generates a log when the contract is created.
@@ -688,7 +700,7 @@ func TestEmptyBlocks(t *testing.T) {
 	defer n.Close()
 
 	commonAncestor := ethservice.BlockChain().CurrentBlock()
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// Setup 10 blocks on the canonical chain
 	setupBlocks(t, ethservice, 10, commonAncestor, func(parent *types.Header) {}, nil, nil)
@@ -814,8 +826,8 @@ func TestTrickRemoteBlockCache(t *testing.T) {
 	}
 	nodeA.Server().AddPeer(nodeB.Server().Self())
 	nodeB.Server().AddPeer(nodeA.Server().Self())
-	apiA := NewConsensusAPI(ethserviceA)
+	apiA := newConsensusAPIWithoutHeartbeat(ethserviceA)
-	apiB := NewConsensusAPI(ethserviceB)
+	apiB := newConsensusAPIWithoutHeartbeat(ethserviceB)
 
 	commonAncestor := ethserviceA.BlockChain().CurrentBlock()
 
@@ -872,7 +884,7 @@ func TestInvalidBloom(t *testing.T) {
 	defer n.Close()
 
 	commonAncestor := ethservice.BlockChain().CurrentBlock()
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// Setup 10 blocks on the canonical chain
 	setupBlocks(t, ethservice, 10, commonAncestor, func(parent *types.Header) {}, nil, nil)
@@ -898,7 +910,7 @@ func TestSimultaneousNewBlock(t *testing.T) {
 	defer n.Close()
 
 	var (
-		api    = NewConsensusAPI(ethservice)
+		api    = newConsensusAPIWithoutHeartbeat(ethservice)
 		parent = preMergeBlocks[len(preMergeBlocks)-1]
 	)
 	for i := 0; i < 10; i++ {
@@ -988,7 +1000,7 @@ func TestWithdrawals(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks)
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// 10: Build Shanghai block with no withdrawals.
 	parent := ethservice.BlockChain().CurrentHeader()
@@ -1105,7 +1117,7 @@ func TestNilWithdrawals(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks)
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 	parent := ethservice.BlockChain().CurrentHeader()
 	aa := common.Address{0xaa}
 
@@ -1301,7 +1313,7 @@ func allBodies(blocks []*types.Block) []*types.Body {
 
 func TestGetBlockBodiesByHash(t *testing.T) {
 	node, eth, blocks := setupBodies(t)
-	api := NewConsensusAPI(eth)
+	api := newConsensusAPIWithoutHeartbeat(eth)
 	defer node.Close()
 
 	tests := []struct {
@@ -1357,7 +1369,7 @@ func TestGetBlockBodiesByHash(t *testing.T) {
 
 func TestGetBlockBodiesByRange(t *testing.T) {
 	node, eth, blocks := setupBodies(t)
-	api := NewConsensusAPI(eth)
+	api := newConsensusAPIWithoutHeartbeat(eth)
 	defer node.Close()
 
 	tests := []struct {
@@ -1438,7 +1450,7 @@ func TestGetBlockBodiesByRange(t *testing.T) {
 
 func TestGetBlockBodiesByRangeInvalidParams(t *testing.T) {
 	node, eth, _ := setupBodies(t)
-	api := NewConsensusAPI(eth)
+	api := newConsensusAPIWithoutHeartbeat(eth)
 	defer node.Close()
 	tests := []struct {
 		start hexutil.Uint64
@@ -1550,7 +1562,7 @@ func TestParentBeaconBlockRoot(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks)
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// 11: Build Shanghai block with no withdrawals.
 	parent := ethservice.BlockChain().CurrentHeader()
@@ -1633,7 +1645,7 @@ func TestWitnessCreationAndConsumption(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, blocks[:9])
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 
 	// Put the 10th block's tx in the pool and produce a new block
 	txs := blocks[9].Transactions()
@@ -1725,7 +1737,7 @@ func TestGetClientVersion(t *testing.T) {
 	n, ethservice := startEthService(t, genesis, preMergeBlocks)
 	defer n.Close()
 
-	api := NewConsensusAPI(ethservice)
+	api := newConsensusAPIWithoutHeartbeat(ethservice)
 	info := engine.ClientVersionV1{
 		Code:    "TT",
 		Name:    "test",
@@ -1799,3 +1811,245 @@ func TestValidateRequests(t *testing.T) {
 		})
 	}
 }
+
+var (
+	testBlobs          []*kzg4844.Blob
+	testBlobCommits    []kzg4844.Commitment
+	testBlobProofs     []kzg4844.Proof
+	testBlobCellProofs [][]kzg4844.Proof
+	testBlobVHashes    [][32]byte
+)
+
+func init() {
+	for i := 0; i < 6; i++ {
+		testBlob := &kzg4844.Blob{byte(i)}
+		testBlobs = append(testBlobs, testBlob)
+
+		testBlobCommit, _ := kzg4844.BlobToCommitment(testBlob)
+		testBlobCommits = append(testBlobCommits, testBlobCommit)
+
+		testBlobProof, _ := kzg4844.ComputeBlobProof(testBlob, testBlobCommit)
+		testBlobProofs = append(testBlobProofs, testBlobProof)
+
+		testBlobCellProof, _ := kzg4844.ComputeCellProofs(testBlob)
+		testBlobCellProofs = append(testBlobCellProofs, testBlobCellProof)
+
+		testBlobVHash := kzg4844.CalcBlobHashV1(sha256.New(), &testBlobCommit)
+		testBlobVHashes = append(testBlobVHashes, testBlobVHash)
+	}
+}
+
+// makeMultiBlobTx is a utility method to construct a random blob tx with
+// certain number of blobs in its sidecar.
+func makeMultiBlobTx(chainConfig *params.ChainConfig, nonce uint64, blobCount int, blobOffset int, key *ecdsa.PrivateKey, version byte) *types.Transaction {
+	var (
+		blobs       []kzg4844.Blob
+		blobHashes  []common.Hash
+		commitments []kzg4844.Commitment
+		proofs      []kzg4844.Proof
+	)
+	for i := 0; i < blobCount; i++ {
+		blobs = append(blobs, *testBlobs[blobOffset+i])
+		commitments = append(commitments, testBlobCommits[blobOffset+i])
+		if version == types.BlobSidecarVersion0 {
+			proofs = append(proofs, testBlobProofs[blobOffset+i])
+		} else {
+			cellProofs, _ := kzg4844.ComputeCellProofs(testBlobs[blobOffset+i])
+			proofs = append(proofs, cellProofs...)
+		}
+		blobHashes = append(blobHashes, testBlobVHashes[blobOffset+i])
+	}
+	blobtx := &types.BlobTx{
+		ChainID:    uint256.MustFromBig(chainConfig.ChainID),
+		Nonce:      nonce,
+		GasTipCap:  uint256.NewInt(1),
+		GasFeeCap:  uint256.NewInt(1000),
+		Gas:        21000,
+		BlobFeeCap: uint256.NewInt(1000),
+		BlobHashes: blobHashes,
+		Value:      uint256.NewInt(100),
+		Sidecar:    types.NewBlobTxSidecar(version, blobs, commitments, proofs),
+	}
+	return types.MustSignNewTx(key, types.LatestSigner(chainConfig), blobtx)
+}
+
+func newGetBlobEnv(t *testing.T, version byte) (*node.Node, *ConsensusAPI) {
+	var (
+		// Create a database pre-initialize with a genesis block
+		config = *params.MergedTestChainConfig
+
+		key1, _ = crypto.GenerateKey()
+		key2, _ = crypto.GenerateKey()
+		key3, _ = crypto.GenerateKey()
+
+		addr1 = crypto.PubkeyToAddress(key1.PublicKey)
+		addr2 = crypto.PubkeyToAddress(key2.PublicKey)
+		addr3 = crypto.PubkeyToAddress(key3.PublicKey)
+	)
+	// Disable Osaka fork for GetBlobsV1
+	if version == 0 {
+		config.OsakaTime = nil
+	}
+	gspec := &core.Genesis{
+		Config: &config,
+		Alloc: types.GenesisAlloc{
+			testAddr: {Balance: testBalance},
+			addr1:    {Balance: testBalance},
+			addr2:    {Balance: testBalance},
+			addr3:    {Balance: testBalance},
+		},
+		Difficulty: common.Big0,
+	}
+	n, ethServ := startEthService(t, gspec, nil)
+
+	// fill blob txs into the pool
+	tx1 := makeMultiBlobTx(&config, 0, 2, 0, key1, version) // blob[0, 2)
+	tx2 := makeMultiBlobTx(&config, 0, 2, 2, key2, version) // blob[2, 4)
+	tx3 := makeMultiBlobTx(&config, 0, 2, 4, key3, version) // blob[4, 6)
+	ethServ.TxPool().Add([]*types.Transaction{tx1, tx2, tx3}, true)
+
+	api := newConsensusAPIWithoutHeartbeat(ethServ)
+	return n, api
+}
+
+func TestGetBlobsV1(t *testing.T) {
+	n, api := newGetBlobEnv(t, 0)
+	defer n.Close()
+
+	suites := []struct {
+		start      int
+		limit      int
+		fillRandom bool
+	}{
+		{
+			start: 0, limit: 1,
+		},
+		{
+			start: 0, limit: 1, fillRandom: true,
+		},
+		{
+			start: 0, limit: 2,
+		},
+		{
+			start: 0, limit: 2, fillRandom: true,
+		},
+		{
+			start: 1, limit: 3,
+		},
+		{
+			start: 1, limit: 3, fillRandom: true,
+		},
+		{
+			start: 0, limit: 6,
+		},
+		{
+			start: 0, limit: 6, fillRandom: true,
+		},
+		{
+			start: 1, limit: 5,
+		},
+		{
+			start: 1, limit: 5, fillRandom: true,
+		},
+	}
+	for i, suite := range suites {
+		// Fill the request for retrieving blobs
+		var (
+			vhashes []common.Hash
+			expect  []*engine.BlobAndProofV1
+		)
+		// fill missing blob at the beginning
+		if suite.fillRandom {
+			vhashes = append(vhashes, testrand.Hash())
+			expect = append(expect, nil)
+		}
+		for j := suite.start; j < suite.limit; j++ {
+			vhashes = append(vhashes, testBlobVHashes[j])
+			expect = append(expect, &engine.BlobAndProofV1{
+				Blob:  testBlobs[j][:],
+				Proof: testBlobProofs[j][:],
+			})
+
+			// fill missing blobs in the middle
+			if suite.fillRandom && rand.Intn(2) == 0 {
+				vhashes = append(vhashes, testrand.Hash())
+				expect = append(expect, nil)
+			}
+		}
+		// fill missing blobs at the end
+		if suite.fillRandom {
+			vhashes = append(vhashes, testrand.Hash())
+			expect = append(expect, nil)
+		}
+		result, err := api.GetBlobsV1(vhashes)
+		if err != nil {
+			t.Errorf("Unexpected error for case %d, %v", i, err)
+		}
+		if !reflect.DeepEqual(result, expect) {
+			t.Fatalf("Unexpected result for case %d", i)
+		}
+	}
+}
+
+func TestGetBlobsV2(t *testing.T) {
+	n, api := newGetBlobEnv(t, 1)
+	defer n.Close()
+
+	suites := []struct {
+		start      int
+		limit      int
+		fillRandom bool
+	}{
+		{
+			start: 0, limit: 1,
+		},
+		{
+			start: 0, limit: 2,
+		},
+		{
+			start: 1, limit: 3,
+		},
+		{
+			start: 0, limit: 6,
+		},
+		{
+			start: 1, limit: 5,
+		},
+		{
+			start: 0, limit: 6, fillRandom: true,
+		},
+	}
+	for i, suite := range suites {
+		// Fill the request for retrieving blobs
+		var (
+			vhashes []common.Hash
+			expect  []*engine.BlobAndProofV2
+		)
+		// fill missing blob
+		if suite.fillRandom {
+			vhashes = append(vhashes, testrand.Hash())
+		}
+		for j := suite.start; j < suite.limit; j++ {
+			vhashes = append(vhashes, testBlobVHashes[j])
+			var cellProofs []hexutil.Bytes
+			for _, proof := range testBlobCellProofs[j] {
+				cellProofs = append(cellProofs, proof[:])
+			}
+			expect = append(expect, &engine.BlobAndProofV2{
+				Blob:       testBlobs[j][:],
+				CellProofs: cellProofs,
+			})
+		}
+		result, err := api.GetBlobsV2(vhashes)
+		if err != nil {
+			t.Errorf("Unexpected error for case %d, %v", i, err)
+		}
+		// null is responded if any blob is missing
+		if suite.fillRandom {
+			expect = nil
+		}
+		if !reflect.DeepEqual(result, expect) {
+			t.Fatalf("Unexpected result for case %d", i)
+		}
+	}
+}