feat: implement reconstruct data column sidecars (#15005)

Author: 0x00101010

beacon-chain/execution/BUILD.bazel

@@ -25,12 +25,14 @@ go_library(
         "//testing/spectest:__subpackages__",
     ],
     deps = [
+        "//beacon-chain/blockchain/kzg:go_default_library",
         "//beacon-chain/cache:go_default_library",
         "//beacon-chain/cache/depositsnapshot:go_default_library",
         "//beacon-chain/core/altair:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
         "//beacon-chain/core/feed/state:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
+        "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/execution/types:go_default_library",
@@ -97,6 +99,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//async/event:go_default_library",
+        "//beacon-chain/blockchain/kzg:go_default_library",
         "//beacon-chain/cache/depositsnapshot:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
         "//beacon-chain/core/feed/state:go_default_library",

beacon-chain/execution/engine_client.go

@@ -13,6 +13,8 @@ import (
 	gethRPC "github.com/ethereum/go-ethereum/rpc"
 	"github.com/holiman/uint256"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/kzg"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/execution/types"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
@@ -44,11 +46,15 @@ var (
 		GetPayloadMethodV3,
 		GetPayloadBodiesByHashV1,
 		GetPayloadBodiesByRangeV1,
+		GetBlobsV1,
 	}
 	electraEngineEndpoints = []string{
 		NewPayloadMethodV4,
 		GetPayloadMethodV4,
 	}
+	fuluEngineEndpoints = []string{
+		GetBlobsV2,
+	}
 )
 
 const (
@@ -85,6 +91,8 @@ const (
 	ExchangeCapabilities = "engine_exchangeCapabilities"
 	// GetBlobsV1 request string for JSON-RPC.
 	GetBlobsV1 = "engine_getBlobsV1"
+	// GetBlobsV2 request string for JSON-RPC.
+	GetBlobsV2 = "engine_getBlobsV2"
 	// Defines the seconds before timing out engine endpoints with non-block execution semantics.
 	defaultEngineTimeout = time.Second
 )
@@ -108,6 +116,7 @@ type Reconstructor interface {
 		ctx context.Context, blindedBlocks []interfaces.ReadOnlySignedBeaconBlock,
 	) ([]interfaces.SignedBeaconBlock, error)
 	ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, hi func(uint64) bool) ([]blocks.VerifiedROBlob, error)
+	ReconstructDataColumnSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte) ([]blocks.VerifiedRODataColumn, error)
 }
 
 // EngineCaller defines a client that can interact with an Ethereum
@@ -302,6 +311,9 @@ func (s *Service) ExchangeCapabilities(ctx context.Context) ([]string, error) {
 	if params.ElectraEnabled() {
 		supportedEngineEndpoints = append(supportedEngineEndpoints, electraEngineEndpoints...)
 	}
+	if params.FuluEnabled() {
+		supportedEngineEndpoints = append(supportedEngineEndpoints, fuluEngineEndpoints...)
+	}
 	var result []string
 	err := s.rpcClient.CallContext(ctx, &result, ExchangeCapabilities, supportedEngineEndpoints)
 	if err != nil {
@@ -495,16 +507,30 @@ func (s *Service) HeaderByNumber(ctx context.Context, number *big.Int) (*types.H
 func (s *Service) GetBlobs(ctx context.Context, versionedHashes []common.Hash) ([]*pb.BlobAndProof, error) {
 	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.GetBlobs")
 	defer span.End()
+
 	// If the execution engine does not support `GetBlobsV1`, return early to prevent encountering an error later.
 	if !s.capabilityCache.has(GetBlobsV1) {
-		return nil, nil
+		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV1))
 	}
 
 	result := make([]*pb.BlobAndProof, len(versionedHashes))
 	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV1, versionedHashes)
 	return result, handleRPCError(err)
 }
 
+func (s *Service) GetBlobsV2(ctx context.Context, versionedHashes []common.Hash) ([]*pb.BlobAndProofV2, error) {
+	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.GetBlobsV2")
+	defer span.End()
+
+	if !s.capabilityCache.has(GetBlobsV2) {
+		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV2))
+	}
+
+	result := make([]*pb.BlobAndProofV2, len(versionedHashes))
+	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV2, versionedHashes)
+	return result, handleRPCError(err)
+}
+
 // ReconstructFullBlock takes in a blinded beacon block and reconstructs
 // a beacon block with a full execution payload via the engine API.
 func (s *Service) ReconstructFullBlock(
@@ -615,6 +641,83 @@ func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.
 	return verifiedBlobs, nil
 }
 
+// ReconstructDataColumnSidecars reconstructs the verified data column sidecars for a given beacon block.
+// It retrieves the KZG commitments from the block body, fetches the associated blobs and cell proofs,
+// and constructs the corresponding verified read-only data column sidecars.
+func (s *Service) ReconstructDataColumnSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte) ([]blocks.VerifiedRODataColumn, error) {
+	blockBody := block.Block().Body()
+	kzgCommitments, err := blockBody.BlobKzgCommitments()
+	if err != nil {
+		return nil, wrapWithBlockRoot(err, blockRoot, "could not get blob KZG commitments")
+	}
+
+	// Collect KZG hashes for all blobs
+	var kzgHashes []common.Hash
+	for _, commitment := range kzgCommitments {
+		kzgHashes = append(kzgHashes, primitives.ConvertKzgCommitmentToVersionedHash(commitment))
+	}
+
+	// Fetch all blobs from EL
+	blobs, err := s.GetBlobsV2(ctx, kzgHashes)
+	if err != nil {
+		return nil, wrapWithBlockRoot(err, blockRoot, "could not get blobs")
+	}
+
+	for _, blobAndCellProofs := range blobs {
+		if blobAndCellProofs == nil {
+			return nil, wrapWithBlockRoot(errors.New("unable to reconstruct data column sidecars, did not get all blobs from EL"), blockRoot, "")
+		}
+	}
+
+	var cellsAndProofs []kzg.CellsAndProofs
+	for _, blobAndCellProofs := range blobs {
+		var blob kzg.Blob
+		copy(blob[:], blobAndCellProofs.Blob)
+		cells, err := kzg.ComputeCells(&blob)
+		if err != nil {
+			return nil, wrapWithBlockRoot(err, blockRoot, "could not compute cells")
+		}
+
+		proofs := make([]kzg.Proof, len(blobAndCellProofs.CellProofs))
+		for i, proof := range blobAndCellProofs.CellProofs {
+			proofs[i] = kzg.Proof(proof)
+		}
+		cellsAndProofs = append(cellsAndProofs, kzg.CellsAndProofs{
+			Cells:  cells,
+			Proofs: proofs,
+		})
+	}
+
+	header, err := block.Header()
+	if err != nil {
+		return nil, wrapWithBlockRoot(err, blockRoot, "could not get header")
+	}
+
+	kzgCommitmentsInclusionProof, err := blocks.MerkleProofKZGCommitments(blockBody)
+	if err != nil {
+		return nil, wrapWithBlockRoot(err, blockRoot, "could not get Merkle proof for KZG commitments")
+	}
+
+	dataColumnSidecars, err := peerdas.DataColumnSidecarsForReconstruct(kzgCommitments, header, kzgCommitmentsInclusionProof, cellsAndProofs)
+	if err != nil {
+		return nil, wrapWithBlockRoot(err, blockRoot, "could not reconstruct data column sidecars")
+	}
+
+	verifiedRODataColumns := make([]blocks.VerifiedRODataColumn, len(dataColumnSidecars))
+	for i, dataColumnSidecar := range dataColumnSidecars {
+		roDataColumn, err := blocks.NewRODataColumnWithRoot(dataColumnSidecar, blockRoot)
+		if err != nil {
+			return nil, wrapWithBlockRoot(err, blockRoot, "new read-only data column with root")
+		}
+
+		verifiedRODataColumns[i] = blocks.NewVerifiedRODataColumn(roDataColumn)
+	}
+
+	log.WithField("root", fmt.Sprintf("%x", blockRoot)).Debug("Data columns reconstructed successfully")
+
+	return verifiedRODataColumns, nil
+}
+
 func fullPayloadFromPayloadBody(
 	header interfaces.ExecutionData, body *pb.ExecutionPayloadBody, bVersion int,
 ) (interfaces.ExecutionData, error) {
@@ -902,3 +1005,8 @@ func toBlockNumArg(number *big.Int) string {
 	}
 	return hexutil.EncodeBig(number)
 }
+
+// wrapWithBlockRoot returns a new error with the given block root.
+func wrapWithBlockRoot(err error, blockRoot [32]byte, message string) error {
+	return errors.Wrap(err, fmt.Sprintf("%s for block %#x", message, blockRoot))
+}

beacon-chain/execution/engine_client_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/holiman/uint256"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/kzg"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filesystem"
 	mocks "github.com/prysmaticlabs/prysm/v5/beacon-chain/execution/testing"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
@@ -2424,7 +2425,7 @@ func TestReconstructBlobSidecars(t *testing.T) {
 	t.Run("get-blobs end point is not supported", func(t *testing.T) {
 		hi := mockSummary(t, []bool{true, true, true, true, true, false})
 		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, hi)
-		require.NoError(t, err)
+		require.ErrorContains(t, "engine_getBlobsV1 is not supported", err)
 		require.Equal(t, 0, len(verifiedBlobs))
 	})
 
@@ -2476,6 +2477,63 @@ func TestReconstructBlobSidecars(t *testing.T) {
 	})
 }
 
+func TestReconstructDataColumnSidecars(t *testing.T) {
+	// Start the trusted setup.
+	err := kzg.Start()
+	require.NoError(t, err)
+
+	// Setup right fork epoch
+	params.SetupTestConfigCleanup(t)
+	cfg := params.BeaconConfig().Copy()
+	cfg.CapellaForkEpoch = 1
+	cfg.DenebForkEpoch = 2
+	cfg.ElectraForkEpoch = 3
+	cfg.FuluForkEpoch = 4
+	params.OverrideBeaconConfig(cfg)
+
+	client := &Service{capabilityCache: &capabilityCache{}}
+	b := util.NewBeaconBlockFulu()
+	b.Block.Slot = 4 * params.BeaconConfig().SlotsPerEpoch
+	kzgCommitments := createRandomKzgCommitments(t, 6)
+	b.Block.Body.BlobKzgCommitments = kzgCommitments
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	sb, err := blocks.NewSignedBeaconBlock(b)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	t.Run("GetBlobsV2 is not supported", func(t *testing.T) {
+		_, err := client.ReconstructDataColumnSidecars(ctx, sb, r)
+		require.ErrorContains(t, "could not get blobs", err)
+	})
+
+	t.Run("receiving all blobs", func(t *testing.T) {
+		blobMasks := []bool{true, true, true, true, true, true}
+		srv := createBlobServerV2(t, 6, blobMasks)
+		defer srv.Close()
+
+		rpcClient, client := setupRpcClientV2(t, srv.URL, client)
+		defer rpcClient.Close()
+
+		dataColumns, err := client.ReconstructDataColumnSidecars(ctx, sb, r)
+		require.NoError(t, err)
+		require.Equal(t, 128, len(dataColumns))
+	})
+
+	t.Run("missing some blobs", func(t *testing.T) {
+		blobMasks := []bool{false, true, true, true, true, true}
+		srv := createBlobServerV2(t, 6, blobMasks)
+		defer srv.Close()
+
+		rpcClient, client := setupRpcClientV2(t, srv.URL, client)
+		defer rpcClient.Close()
+
+		dataColumns, err := client.ReconstructDataColumnSidecars(ctx, sb, r)
+		require.ErrorContains(t, "unable to reconstruct data column sidecars, did not get all blobs from EL", err)
+		require.Equal(t, 0, len(dataColumns))
+	})
+}
+
 func createRandomKzgCommitments(t *testing.T, num int) [][]byte {
 	kzgCommitments := make([][]byte, num)
 	for i := range kzgCommitments {
@@ -2511,6 +2569,41 @@ func createBlobServer(t *testing.T, numBlobs int, callbackFuncs ...func()) *http
 	}))
 }
 
+func createBlobServerV2(t *testing.T, numBlobs int, blobMasks []bool) *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		defer func() {
+			require.NoError(t, r.Body.Close())
+		}()
+
+		require.Equal(t, len(blobMasks), numBlobs)
+
+		blobAndCellProofs := make([]*pb.BlobAndCellProofJson, numBlobs)
+		for i := range blobAndCellProofs {
+			if !blobMasks[i] {
+				continue
+			}
+
+			blobAndCellProofs[i] = &pb.BlobAndCellProofJson{
+				Blob:       []byte("0xblob"),
+				CellProofs: []hexutil.Bytes{},
+			}
+			for j := 0; j < int(params.BeaconConfig().NumberOfColumns); j++ {
+				blobAndCellProofs[i].CellProofs = append(blobAndCellProofs[i].CellProofs, []byte(fmt.Sprintf("0xproof%d", j)))
+			}
+		}
+
+		respJSON := map[string]interface{}{
+			"jsonrpc": "2.0",
+			"id":      1,
+			"result":  blobAndCellProofs,
+		}
+
+		err := json.NewEncoder(w).Encode(respJSON)
+		require.NoError(t, err)
+	}))
+}
+
 func setupRpcClient(t *testing.T, url string, client *Service) (*rpc.Client, *Service) {
 	rpcClient, err := rpc.DialHTTP(url)
 	require.NoError(t, err)
@@ -2522,6 +2615,12 @@ func setupRpcClient(t *testing.T, url string, client *Service) (*rpc.Client, *Se
 	return rpcClient, client
 }
 
+func setupRpcClientV2(t *testing.T, url string, client *Service) (*rpc.Client, *Service) {
+	rpcClient, client := setupRpcClient(t, url, client)
+	client.capabilityCache = &capabilityCache{capabilities: map[string]interface{}{GetBlobsV2: nil}}
+	return rpcClient, client
+}
+
 func testNewBlobVerifier() verification.NewBlobVerifier {
 	return func(b blocks.ROBlob, reqs []verification.Requirement) verification.BlobVerifier {
 		return &verification.MockBlobVerifier{

beacon-chain/execution/testing/mock_engine_client.go

@@ -113,6 +113,10 @@ func (e *EngineClient) ReconstructBlobSidecars(context.Context, interfaces.ReadO
 	return e.BlobSidecars, e.ErrorBlobSidecars
 }
 
+func (e *EngineClient) ReconstructDataColumnSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte) ([]blocks.VerifiedRODataColumn, error) {
+	return nil, nil
+}
+
 // GetTerminalBlockHash --
 func (e *EngineClient) GetTerminalBlockHash(ctx context.Context, transitionTime uint64) ([]byte, bool, error) {
 	ttd := new(big.Int)

proto/engine/v1/json_marshal_unmarshal.go

@@ -1282,3 +1282,30 @@ func (b *BlobAndProof) UnmarshalJSON(enc []byte) error {
 
 	return nil
 }
+
+type BlobAndCellProofJson struct {
+	Blob       hexutil.Bytes   `json:"blob"`
+	CellProofs []hexutil.Bytes `json:"cell_proofs"`
+}
+
+// UnmarshalJSON implements the json unmarshaler interface for BlobAndCellProofJson.
+func (b *BlobAndProofV2) UnmarshalJSON(enc []byte) error {
+	var dec *BlobAndCellProofJson
+	if err := json.Unmarshal(enc, &dec); err != nil {
+		return err
+	}
+
+	blob := make([]byte, fieldparams.BlobLength)
+	copy(blob, dec.Blob)
+	b.Blob = blob
+
+	cellProofs := make([][]byte, len(dec.CellProofs))
+	for i, proof := range dec.CellProofs {
+		p := make([]byte, fieldparams.BLSPubkeyLength)
+		copy(p, proof)
+		cellProofs[i] = p
+	}
+	b.CellProofs = cellProofs
+
+	return nil
+}