Finish ReconstructDataColumnSidecars

Author: 0x00101010

Diff for: 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, hi func(uint64) bool) ([]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 {
@@ -492,19 +504,38 @@ func (s *Service) HeaderByNumber(ctx context.Context, number *big.Int) (*types.H
 }
 
 // GetBlobs returns the blob and proof from the execution engine for the given versioned hashes.
-func (s *Service) GetBlobs(ctx context.Context, versionedHashes []common.Hash) ([]*pb.BlobAndProof, error) {
+func (s *Service) GetBlobs(ctx context.Context, slot primitives.Slot, versionedHashes []common.Hash) (any, 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) {
+
+	method, proto := s.getPayloadMethodAndMessage(slot)
+	// If the execution engine does not support the method, return early to prevent encountering an error later.
+	if !s.capabilityCache.has(method) {
 		return nil, nil
 	}
 
-	result := make([]*pb.BlobAndProof, len(versionedHashes))
-	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV1, versionedHashes)
+	var result any
+	switch proto.(type) {
+	case *pb.BlobAndProofV2:
+		result = make([]*pb.BlobAndProofV2, len(versionedHashes))
+	case *pb.BlobAndProof:
+		result = make([]*pb.BlobAndProof, len(versionedHashes))
+	default:
+		return nil, errors.New("unsupported blob proof type")
+	}
+
+	err := s.rpcClient.CallContext(ctx, &result, method, versionedHashes)
 	return result, handleRPCError(err)
 }
 
+func (s *Service) getPayloadMethodAndMessage(slot primitives.Slot) (string, proto.Message) {
+	pe := slots.ToEpoch(slot)
+	if pe >= params.BeaconConfig().FuluForkEpoch {
+		return GetBlobsV2, &pb.BlobAndProofV2{}
+	}
+	return GetBlobsV1, &pb.BlobAndProof{}
+}
+
 // 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(
@@ -561,10 +592,14 @@ func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.
 	}
 
 	// Fetch blobs from EL
-	blobs, err := s.GetBlobs(ctx, kzgHashes)
+	result, err := s.GetBlobs(ctx, block.Block().Slot(), kzgHashes)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get blobs")
 	}
+	blobs, ok := result.([]*pb.BlobAndProof)
+	if !ok {
+		return nil, errors.New("invalid blob proof type")
+	}
 	if len(blobs) == 0 {
 		return nil, nil
 	}
@@ -615,6 +650,88 @@ 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, errors.Wrap(err, "could not get blob KZG commitments")
+	}
+
+	// Collect KZG hashes for all blobs
+	var kzgHashes []common.Hash
+	var kzgIndexes []int
+	for i, commitment := range kzgCommitments {
+		kzgHashes = append(kzgHashes, primitives.ConvertKzgCommitmentToVersionedHash(commitment))
+		kzgIndexes = append(kzgIndexes, i)
+	}
+
+	// Fetch all blobs from EL
+	result, err := s.GetBlobs(ctx, block.Block().Slot(), kzgHashes)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get blobs")
+	}
+
+	blobAndProofs, ok := result.([]*pb.BlobAndProofV2)
+	if !ok {
+		return nil, errors.New("invalid blob and proof type")
+	}
+	for _, blobAndCellProofs := range blobAndProofs {
+		if blobAndCellProofs == nil {
+			return nil, errors.Wrap(err, "unable to reconstruct data column sidecars, not enough blob data from EL")
+		}
+	}
+
+	var cellsAndProofs []kzg.CellsAndProofs
+	for _, blobAndCellProofs := range blobAndProofs {
+		blob := kzg.Blob(blobAndCellProofs.Blob)
+		cells, err := kzg.ComputeCells(&blob)
+		if err != nil {
+			return nil, errors.Wrap(err, "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, errors.Wrap(err, "could not get header")
+	}
+
+	kzgCommitmentsInclusionProof, err := blocks.MerkleProofKZGCommitments(blockBody)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get Merkle proof for KZG commitments")
+	}
+
+	dataColumnSidecars, err := peerdas.DataColumnSidecarsForReconstruct(kzgCommitments, header, kzgCommitmentsInclusionProof, cellsAndProofs)
+	if err != nil {
+		return nil, errors.Wrap(err, "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, errors.Wrap(err, "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) {