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/epbs: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 {

beacon-chain/blockchain/receive_execution_payload_envelope.go

@@ -0,0 +1,162 @@
+package blockchain
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/epbs"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/das"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/execution"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	"github.com/sirupsen/logrus"
+	"go.opencensus.io/trace"
+	"golang.org/x/sync/errgroup"
+)
+
+// ReceiveExecutionPayloadEnvelope is a function that defines the operations (minus pubsub)
+// that are performed on a received execution payload envelope. The operations consist of:
+//  1. Validate the payload, apply state transition.
+//  2. Apply fork choice to the processed payload
+//  3. Save latest head info
+func (s *Service) ReceiveExecutionPayloadEnvelope(ctx context.Context, envelope interfaces.ROExecutionPayloadEnvelope, _ das.AvailabilityStore) error {
+	receivedTime := time.Now()
+	root, err := envelope.BeaconBlockRoot()
+	if err != nil {
+		return errors.Wrap(err, "could not get beacon block root")
+	}
+	s.payloadBeingSynced.set(root)
+	defer s.payloadBeingSynced.unset(root)
+
+	preState, err := s.getPayloadEnvelopePrestate(ctx, envelope)
+	if err != nil {
+		return errors.Wrap(err, "could not get prestate")
+	}
+
+	eg, _ := errgroup.WithContext(ctx)
+	var postState state.BeaconState
+	eg.Go(func() error {
+		if err := epbs.ValidatePayloadStateTransition(ctx, preState, envelope); err != nil {
+			return errors.Wrap(err, "failed to validate consensus state transition function")
+		}
+		return nil
+	})
+	var isValidPayload bool
+	eg.Go(func() error {
+		var err error
+		isValidPayload, err = s.validateExecutionOnEnvelope(ctx, envelope, root)
+		if err != nil {
+			return errors.Wrap(err, "could not notify the engine of the new payload")
+		}
+		return nil
+	})
+
+	if err := eg.Wait(); err != nil {
+		return err
+	}
+	_ = isValidPayload
+	_ = postState
+	daStartTime := time.Now()
+	// TODO: Add DA check
+	daWaitedTime := time.Since(daStartTime)
+	dataAvailWaitedTime.Observe(float64(daWaitedTime.Milliseconds()))
+	// TODO: Add Head update, cache handling, postProcessing
+	timeWithoutDaWait := time.Since(receivedTime) - daWaitedTime
+	executionEngineProcessingTime.Observe(float64(timeWithoutDaWait.Milliseconds()))
+	return nil
+}
+
+// notifyNewPayload signals execution engine on a new payload.
+// It returns true if the EL has returned VALID for the block
+func (s *Service) notifyNewEnvelope(ctx context.Context, envelope interfaces.ROExecutionPayloadEnvelope) (bool, error) {
+	ctx, span := trace.StartSpan(ctx, "blockChain.notifyNewPayload")
+	defer span.End()
+
+	payload, err := envelope.Execution()
+	if err != nil {
+		return false, errors.Wrap(err, "could not get execution payload")
+	}
+
+	var lastValidHash []byte
+	var versionedHashes []common.Hash
+	versionedHashes, err = envelope.VersionedHashes()
+	if err != nil {
+		return false, errors.Wrap(err, "could not get versioned hashes to feed the engine")
+	}
+	root, err := envelope.BeaconBlockRoot()
+	if err != nil {
+		return false, errors.Wrap(err, "could not get beacon block root")
+	}
+	parentRoot, err := s.ParentRoot(root)
+	if err != nil {
+		return false, errors.Wrap(err, "could not get parent block root")
+	}
+	pr := common.Hash(parentRoot)
+	lastValidHash, err = s.cfg.ExecutionEngineCaller.NewPayload(ctx, payload, versionedHashes, &pr)
+	switch {
+	case err == nil:
+		newPayloadValidNodeCount.Inc()
+		return true, nil
+	case errors.Is(err, execution.ErrAcceptedSyncingPayloadStatus):
+		newPayloadOptimisticNodeCount.Inc()
+		log.WithFields(logrus.Fields{
+			"payloadBlockHash": fmt.Sprintf("%#x", bytesutil.Trunc(payload.BlockHash())),
+		}).Info("Called new payload with optimistic block")
+		return false, nil
+	case errors.Is(err, execution.ErrInvalidPayloadStatus):
+		lvh := bytesutil.ToBytes32(lastValidHash)
+		return false, invalidBlock{
+			error:         ErrInvalidPayload,
+			lastValidHash: lvh,
+		}
+	default:
+		return false, errors.WithMessage(ErrUndefinedExecutionEngineError, err.Error())
+	}
+}
+
+// validateExecutionOnEnvelope notifies the engine of the incoming execution payload and returns true if the payload is valid
+func (s *Service) validateExecutionOnEnvelope(ctx context.Context, e interfaces.ROExecutionPayloadEnvelope, parentRoot [32]byte) (bool, error) {
+	isValidPayload, err := s.notifyNewEnvelope(ctx, e)
+	if err != nil {
+		blockRoot, rootErr := e.BeaconBlockRoot()
+		if rootErr != nil {
+			return false, err
+		}
+		s.cfg.ForkChoiceStore.Lock()
+		err = s.handleInvalidExecutionError(ctx, err, blockRoot, parentRoot)
+		s.cfg.ForkChoiceStore.Unlock()
+		return false, err
+	}
+	return isValidPayload, nil
+}
+
+func (s *Service) getPayloadEnvelopePrestate(ctx context.Context, e interfaces.ROExecutionPayloadEnvelope) (state.BeaconState, error) {
+	ctx, span := trace.StartSpan(ctx, "blockChain.getPayloadEnvelopePreState")
+	defer span.End()
+
+	// Verify incoming payload has a valid pre state.
+	root, err := e.BeaconBlockRoot()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get beacon block root")
+	}
+	// Verify the referred block is known to forkchoice
+	if !s.InForkchoice(root) {
+		return nil, errors.New("Cannot import execution payload envelope for unknown block")
+	}
+	if err := s.verifyBlkPreState(ctx, root); err != nil {
+		return nil, errors.Wrap(err, "could not verify payload prestate")
+	}
+
+	preState, err := s.cfg.StateGen.StateByRoot(ctx, root)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get pre state")
+	}
+	if preState == nil || preState.IsNil() {
+		return nil, errors.Wrap(err, "nil pre state")
+	}
+	return preState, nil
+}

beacon-chain/blockchain/service.go

@@ -65,6 +65,7 @@ type Service struct {
 	syncComplete                  chan struct{}
 	blobNotifiers                 *blobNotifierMap
 	blockBeingSynced              *currentlySyncingBlock
+	payloadBeingSynced            *currentlySyncingPayload
 	blobStorage                   *filesystem.BlobStorage
 	lastPublishedLightClientEpoch primitives.Epoch
 }
@@ -544,7 +545,7 @@ func (s *Service) saveGenesisData(ctx context.Context, genesisState state.Beacon
 // 2.) Check DB.
 // Checking 1.) is ten times faster than checking 2.)
 // this function requires a lock in forkchoice
-func (s *Service) hasBlock(ctx context.Context, root [32]byte) bool {
+func (s *Service) chainHasBlock(ctx context.Context, root [32]byte) bool {
 	if s.cfg.ForkChoiceStore.HasNode(root) {
 		return true
 	}

beacon-chain/blockchain/service_test.go

@@ -382,8 +382,8 @@ func TestHasBlock_ForkChoiceAndDB_DoublyLinkedTree(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, s.cfg.ForkChoiceStore.InsertNode(ctx, beaconState, r))
 
-	assert.Equal(t, false, s.hasBlock(ctx, [32]byte{}), "Should not have block")
-	assert.Equal(t, true, s.hasBlock(ctx, r), "Should have block")
+	assert.Equal(t, false, s.chainHasBlock(ctx, [32]byte{}), "Should not have block")
+	assert.Equal(t, true, s.chainHasBlock(ctx, r), "Should have block")
 }
 
 func TestServiceStop_SaveCachedBlocks(t *testing.T) {