Process Execution Payload Envelope in Chain Service

beacon-chain/blockchain/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "chain_info.go",
         "chain_info_forkchoice.go",
         "currently_syncing_block.go",
+        "currently_syncing_execution_payload_envelope.go",
         "defragment.go",
         "error.go",
         "execution_engine.go",
@@ -26,6 +27,7 @@ go_library(
         "receive_attestation.go",
         "receive_blob.go",
         "receive_block.go",
+        "receive_execution_payload_envelope.go",
         "service.go",
         "tracked_proposer.go",
         "weak_subjectivity_checks.go",
@@ -44,6 +46,7 @@ go_library(
         "//beacon-chain/cache:go_default_library",
         "//beacon-chain/core/altair:go_default_library",
         "//beacon-chain/core/blocks:go_default_library",
+        "//beacon-chain/core/electra:go_default_library",
         "//beacon-chain/core/epoch/precompute:go_default_library",
         "//beacon-chain/core/feed:go_default_library",
         "//beacon-chain/core/feed/state:go_default_library",

beacon-chain/blockchain/chain_info.go

@@ -530,6 +530,11 @@ func (s *Service) recoverStateSummary(ctx context.Context, blockRoot [32]byte) (
 	return nil, errBlockDoesNotExist
 }
 
+// PayloadBeingSynced returns whether the payload for the block with the given root is currently being synced
+func (s *Service) PayloadBeingSynced(root [32]byte) bool {
+	return s.payloadBeingSynced.isSyncing(root)
+}
+
 // BlockBeingSynced returns whether the block with the given root is currently being synced
 func (s *Service) BlockBeingSynced(root [32]byte) bool {
 	return s.blockBeingSynced.isSyncing(root)

beacon-chain/blockchain/currently_syncing_execution_payload_envelope.go

@@ -0,0 +1,27 @@
+package blockchain
+
+import "sync"
+
+type currentlySyncingPayload struct {
+	sync.Mutex
+	roots map[[32]byte]struct{}
+}
+
+func (b *currentlySyncingPayload) set(root [32]byte) {
+	b.Lock()
+	defer b.Unlock()
+	b.roots[root] = struct{}{}
+}
+
+func (b *currentlySyncingPayload) unset(root [32]byte) {
+	b.Lock()
+	defer b.Unlock()
+	delete(b.roots, root)
+}
+
+func (b *currentlySyncingPayload) isSyncing(root [32]byte) bool {
+	b.Lock()
+	defer b.Unlock()
+	_, ok := b.roots[root]
+	return ok
+}

beacon-chain/blockchain/execution_engine.go

@@ -2,7 +2,6 @@ package blockchain
 
 import (
 	"context"
-	"crypto/sha256"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -28,8 +27,6 @@ import (
 	"go.opencensus.io/trace"
 )
 
-const blobCommitmentVersionKZG uint8 = 0x01
-
 var defaultLatestValidHash = bytesutil.PadTo([]byte{0xff}, 32)
 
 // notifyForkchoiceUpdate signals execution engine the fork choice updates. Execution engine should:
@@ -402,13 +399,7 @@ func kzgCommitmentsToVersionedHashes(body interfaces.ReadOnlyBeaconBlockBody) ([
 
 	versionedHashes := make([]common.Hash, len(commitments))
 	for i, commitment := range commitments {
-		versionedHashes[i] = ConvertKzgCommitmentToVersionedHash(commitment)
+		versionedHashes[i] = primitives.ConvertKzgCommitmentToVersionedHash(commitment)
 	}
 	return versionedHashes, nil
 }
-
-func ConvertKzgCommitmentToVersionedHash(commitment []byte) common.Hash {
-	versionedHash := sha256.Sum256(commitment)
-	versionedHash[0] = blobCommitmentVersionKZG
-	return versionedHash
-}

beacon-chain/blockchain/execution_engine_test.go

@@ -1056,8 +1056,8 @@ func TestService_removeInvalidBlockAndState(t *testing.T) {
 
 	require.NoError(t, service.removeInvalidBlockAndState(ctx, [][32]byte{r1, r2}))
 
-	require.Equal(t, false, service.hasBlock(ctx, r1))
-	require.Equal(t, false, service.hasBlock(ctx, r2))
+	require.Equal(t, false, service.chainHasBlock(ctx, r1))
+	require.Equal(t, false, service.chainHasBlock(ctx, r2))
 	require.Equal(t, false, service.cfg.BeaconDB.HasStateSummary(ctx, r1))
 	require.Equal(t, false, service.cfg.BeaconDB.HasStateSummary(ctx, r2))
 	has, err := service.cfg.StateGen.HasState(ctx, r1)

beacon-chain/blockchain/log.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/blocks"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	consensus_types "github.com/prysmaticlabs/prysm/v5/consensus-types"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
@@ -111,18 +110,7 @@ func logBlockSyncStatus(block interfaces.ReadOnlyBeaconBlock, blockRoot [32]byte
 }
 
 // logs payload related data every slot.
-func logPayload(block interfaces.ReadOnlyBeaconBlock) error {
-	isExecutionBlk, err := blocks.IsExecutionBlock(block.Body())
-	if err != nil {
-		return errors.Wrap(err, "could not determine if block is execution block")
-	}
-	if !isExecutionBlk {
-		return nil
-	}
-	payload, err := block.Body().Execution()
-	if err != nil {
-		return err
-	}
+func logPayload(ver int, payload interfaces.ExecutionData) error {
 	if payload.GasLimit() == 0 {
 		return errors.New("gas limit should not be 0")
 	}
@@ -133,17 +121,12 @@ func logPayload(block interfaces.ReadOnlyBeaconBlock) error {
 		"blockNumber": payload.BlockNumber(),
 		"gasUtilized": fmt.Sprintf("%.2f", gasUtilized),
 	}
-	if block.Version() >= version.Capella {
+	if ver >= version.Capella {
 		withdrawals, err := payload.Withdrawals()
 		if err != nil {
 			return errors.Wrap(err, "could not get withdrawals")
 		}
 		fields["withdrawals"] = len(withdrawals)
-		changes, err := block.Body().BLSToExecutionChanges()
-		if err != nil {
-			return errors.Wrap(err, "could not get BLSToExecutionChanges")
-		}
-		fields["blsToExecutionChanges"] = len(changes)
 	}
 	log.WithFields(fields).Debug("Synced new payload")
 	return nil

beacon-chain/blockchain/metrics.go

@@ -182,6 +182,10 @@ var (
 		Name: "chain_service_processing_milliseconds",
 		Help: "Total time to call a chain service in ReceiveBlock()",
 	})
+	executionEngineProcessingTime = promauto.NewSummary(prometheus.SummaryOpts{
+		Name: "execution_engine_processing_milliseconds",
+		Help: "Total time to process an execution payload envelope in ReceiveExecutionPayloadEnvelope()",
+	})
 	dataAvailWaitedTime = promauto.NewSummary(prometheus.SummaryOpts{
 		Name: "da_waited_time_milliseconds",
 		Help: "Total time spent waiting for a data availability check in ReceiveBlock()",

beacon-chain/blockchain/receive_attestation.go

@@ -176,7 +176,7 @@ func (s *Service) processAttestations(ctx context.Context, disparity time.Durati
 		}
 
 		hasState := s.cfg.BeaconDB.HasStateSummary(ctx, bytesutil.ToBytes32(a.GetData().BeaconBlockRoot))
-		hasBlock := s.hasBlock(ctx, bytesutil.ToBytes32(a.GetData().BeaconBlockRoot))
+		hasBlock := s.chainHasBlock(ctx, bytesutil.ToBytes32(a.GetData().BeaconBlockRoot))
 		if !(hasState && hasBlock) {
 			continue
 		}

beacon-chain/blockchain/receive_block.go

@@ -206,8 +206,14 @@ func (s *Service) ReceiveBlock(ctx context.Context, block interfaces.ReadOnlySig
 		log.WithError(err).Error("Unable to log block sync status")
 	}
 	// Log payload data
-	if err := logPayload(blockCopy.Block()); err != nil {
-		log.WithError(err).Error("Unable to log debug block payload data")
+	ver := blockCopy.Version()
+	if ver >= version.Bellatrix {
+		payload, err := blockCopy.Block().Body().Execution()
+		if err != nil {
+			log.WithError(err).Error("unable to get execution payload")
+		} else if err := logPayload(ver, payload); err != nil {
+			log.WithError(err).Error("Unable to log debug block payload data")
+		}
 	}
 	// Log state transition data.
 	if err := logStateTransitionData(blockCopy.Block()); err != nil {