Local proposer end to end

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",
@@ -25,6 +26,8 @@ go_library(
         "receive_attestation.go",
         "receive_blob.go",
         "receive_block.go",
+        "receive_execution_payload_envelope.go",
+        "receive_payload_attestation_message.go",
         "service.go",
         "tracked_proposer.go",
         "weak_subjectivity_checks.go",
@@ -43,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",
@@ -97,6 +101,7 @@ go_library(
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
+        "@io_opencensus_go//trace:go_default_library",
         "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )
@@ -109,6 +114,7 @@ go_test(
         "chain_info_norace_test.go",
         "chain_info_test.go",
         "checktags_test.go",
+        "epbs_test.go",
         "error_test.go",
         "execution_engine_test.go",
         "forkchoice_update_execution_test.go",
@@ -124,6 +130,7 @@ go_test(
         "process_block_test.go",
         "receive_attestation_test.go",
         "receive_block_test.go",
+        "receive_execution_payload_envelope_test.go",
         "service_norace_test.go",
         "service_test.go",
         "setup_test.go",
@@ -177,6 +184,7 @@ go_test(
         "//testing/assert:go_default_library",
         "//testing/require:go_default_library",
         "//testing/util:go_default_library",
+        "//testing/util/random:go_default_library",
         "//time:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",

beacon-chain/blockchain/chain_info.go

@@ -42,14 +42,15 @@ type ForkchoiceFetcher interface {
 	GetProposerHead() [32]byte
 	SetForkChoiceGenesisTime(uint64)
 	UpdateHead(context.Context, primitives.Slot)
-	HighestReceivedBlockSlot() primitives.Slot
+	HighestReceivedBlockSlotRoot() (primitives.Slot, [32]byte)
 	ReceivedBlocksLastEpoch() (uint64, error)
 	InsertNode(context.Context, state.BeaconState, [32]byte) error
 	ForkChoiceDump(context.Context) (*forkchoice.Dump, error)
 	NewSlot(context.Context, primitives.Slot) error
 	ProposerBoost() [32]byte
 	RecentBlockSlot(root [32]byte) (primitives.Slot, error)
 	IsCanonical(ctx context.Context, blockRoot [32]byte) (bool, error)
+	GetPTCVote(root [32]byte) primitives.PTCStatus
 }
 
 // TimeFetcher retrieves the Ethereum consensus data that's related to time.
@@ -118,6 +119,12 @@ type OptimisticModeFetcher interface {
 	IsOptimisticForRoot(ctx context.Context, root [32]byte) (bool, error)
 }
 
+// ExecutionPayloadFetcher defines a common interface that returns forkchoice
+// information about payload block hashes
+type ExecutionPayloadFetcher interface {
+	HashInForkchoice([32]byte) bool
+}
+
 // FinalizedCheckpt returns the latest finalized checkpoint from chain store.
 func (s *Service) FinalizedCheckpt() *ethpb.Checkpoint {
 	s.cfg.ForkChoiceStore.RLock()
@@ -399,6 +406,14 @@ func (s *Service) InForkchoice(root [32]byte) bool {
 	return s.cfg.ForkChoiceStore.HasNode(root)
 }
 
+// HashInForkchoice returns true if the given payload block hash is found in
+// forkchoice
+func (s *Service) HashInForkchoice(hash [32]byte) bool {
+	s.cfg.ForkChoiceStore.RLock()
+	defer s.cfg.ForkChoiceStore.RUnlock()
+	return s.cfg.ForkChoiceStore.HasHash(hash)
+}
+
 // IsOptimisticForRoot takes the root as argument instead of the current head
 // and returns true if it is optimistic.
 func (s *Service) IsOptimisticForRoot(ctx context.Context, root [32]byte) (bool, error) {

beacon-chain/blockchain/chain_info_forkchoice.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/forkchoice"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 )
 
@@ -29,11 +30,11 @@ func (s *Service) SetForkChoiceGenesisTime(timestamp uint64) {
 	s.cfg.ForkChoiceStore.SetGenesisTime(timestamp)
 }
 
-// HighestReceivedBlockSlot returns the corresponding value from forkchoice
-func (s *Service) HighestReceivedBlockSlot() primitives.Slot {
+// HighestReceivedBlockSlotRoot returns the corresponding value from forkchoice
+func (s *Service) HighestReceivedBlockSlotRoot() (primitives.Slot, [32]byte) {
 	s.cfg.ForkChoiceStore.RLock()
 	defer s.cfg.ForkChoiceStore.RUnlock()
-	return s.cfg.ForkChoiceStore.HighestReceivedBlockSlot()
+	return s.cfg.ForkChoiceStore.HighestReceivedBlockSlotRoot()
 }
 
 // ReceivedBlocksLastEpoch returns the corresponding value from forkchoice
@@ -99,3 +100,30 @@ func (s *Service) ParentRoot(root [32]byte) ([32]byte, error) {
 	defer s.cfg.ForkChoiceStore.RUnlock()
 	return s.cfg.ForkChoiceStore.ParentRoot(root)
 }
+
+// GetPTCVote wraps a call to the corresponding method in forkchoice and checks
+// the currently syncing status
+// Warning: this method will return the current PTC status regardless of
+// timeliness. A client MUST call this method when about to submit a PTC
+// attestation, that is exactly at the threshold to submit the attestation.
+func (s *Service) GetPTCVote(root [32]byte) primitives.PTCStatus {
+	s.cfg.ForkChoiceStore.RLock()
+	f := s.cfg.ForkChoiceStore.GetPTCVote()
+	s.cfg.ForkChoiceStore.RUnlock()
+	if f != primitives.PAYLOAD_ABSENT {
+		return f
+	}
+	f, isSyncing := s.payloadBeingSynced.isSyncing(root)
+	if isSyncing {
+		return f
+	}
+	return primitives.PAYLOAD_ABSENT
+}
+
+// insertPayloadEnvelope wraps a locked call to the corresponding method in
+// forkchoice
+func (s *Service) insertPayloadEnvelope(envelope interfaces.ROExecutionPayloadEnvelope) error {
+	s.cfg.ForkChoiceStore.Lock()
+	defer s.cfg.ForkChoiceStore.Unlock()
+	return s.cfg.ForkChoiceStore.InsertPayloadEnvelope(envelope)
+}

beacon-chain/blockchain/currently_syncing_execution_payload_envelope.go

@@ -0,0 +1,36 @@
+package blockchain
+
+import (
+	"sync"
+
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+)
+
+type currentlySyncingPayload struct {
+	sync.Mutex
+	roots map[[32]byte]primitives.PTCStatus
+}
+
+func (b *currentlySyncingPayload) set(envelope interfaces.ROExecutionPayloadEnvelope) {
+	b.Lock()
+	defer b.Unlock()
+	if envelope.PayloadWithheld() {
+		b.roots[envelope.BeaconBlockRoot()] = primitives.PAYLOAD_WITHHELD
+	} else {
+		b.roots[envelope.BeaconBlockRoot()] = primitives.PAYLOAD_PRESENT
+	}
+}
+
+func (b *currentlySyncingPayload) unset(root [32]byte) {
+	b.Lock()
+	defer b.Unlock()
+	delete(b.roots, root)
+}
+
+func (b *currentlySyncingPayload) isSyncing(root [32]byte) (status primitives.PTCStatus, isSyncing bool) {
+	b.Lock()
+	defer b.Unlock()
+	status, isSyncing = b.roots[root]
+	return
+}

beacon-chain/blockchain/epbs_test.go

@@ -0,0 +1,18 @@
+package blockchain
+
+import (
+	"testing"
+
+	doublylinkedtree "github.com/prysmaticlabs/prysm/v5/beacon-chain/forkchoice/doubly-linked-tree"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+)
+
+func TestServiceGetPTCVote(t *testing.T) {
+	c := &currentlySyncingPayload{roots: make(map[[32]byte]primitives.PTCStatus)}
+	s := &Service{cfg: &config{ForkChoiceStore: doublylinkedtree.New()}, payloadBeingSynced: c}
+	r := [32]byte{'r'}
+	require.Equal(t, primitives.PAYLOAD_ABSENT, s.GetPTCVote(r))
+	c.roots[r] = primitives.PAYLOAD_WITHHELD
+	require.Equal(t, primitives.PAYLOAD_WITHHELD, s.GetPTCVote(r))
+}

beacon-chain/blockchain/error.go

@@ -30,6 +30,9 @@ var (
 	ErrNotCheckpoint = errors.New("not a checkpoint in forkchoice")
 	// ErrNilHead is returned when no head is present in the blockchain service.
 	ErrNilHead = errors.New("nil head")
+	// errInvalidValidatorIndex is returned when a validator index is
+	// invalid or unexpected
+	errInvalidValidatorIndex = errors.New("invalid validator index")
 )
 
 var errMaxBlobsExceeded = errors.New("Expected commitments in block exceeds MAX_BLOBS_PER_BLOCK")

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 (
 	"github.com/sirupsen/logrus"
 )
 
-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/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/options.go

@@ -1,6 +1,8 @@
 package blockchain
 
 import (
+	"sync"
+
 	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
@@ -69,6 +71,22 @@ func WithDepositCache(c cache.DepositCache) Option {
 	}
 }
 
+// WithPayloadAttestationCache for payload attestation cache.
+func WithPayloadAttestationCache(c *cache.PayloadAttestationCache) Option {
+	return func(s *Service) error {
+		s.cfg.PayloadAttestationCache = c
+		return nil
+	}
+}
+
+// WithPayloadEnvelopeCache for payload envelope cache.
+func WithPayloadEnvelopeCache(c *sync.Map) Option {
+	return func(s *Service) error {
+		s.cfg.PayloadEnvelopeCache = c
+		return nil
+	}
+}
+
 // WithPayloadIDCache for payload ID cache.
 func WithPayloadIDCache(c *cache.PayloadIDCache) Option {
 	return func(s *Service) error {

beacon-chain/blockchain/process_attestation.go

@@ -97,7 +97,7 @@ func (s *Service) OnAttestation(ctx context.Context, a ethpb.Att, disparity time
 	// We assume trusted attestation in this function has verified signature.
 
 	// Update forkchoice store with the new attestation for updating weight.
-	s.cfg.ForkChoiceStore.ProcessAttestation(ctx, indexedAtt.GetAttestingIndices(), bytesutil.ToBytes32(a.GetData().BeaconBlockRoot), a.GetData().Target.Epoch)
+	s.cfg.ForkChoiceStore.ProcessAttestation(ctx, indexedAtt.GetAttestingIndices(), bytesutil.ToBytes32(a.GetData().BeaconBlockRoot), a.GetData().Slot)
 
 	return nil
 }

beacon-chain/blockchain/process_block.go

@@ -375,7 +375,7 @@ func (s *Service) handleBlockAttestations(ctx context.Context, blk interfaces.Re
 		}
 		r := bytesutil.ToBytes32(a.GetData().BeaconBlockRoot)
 		if s.cfg.ForkChoiceStore.HasNode(r) {
-			s.cfg.ForkChoiceStore.ProcessAttestation(ctx, indices, r, a.GetData().Target.Epoch)
+			s.cfg.ForkChoiceStore.ProcessAttestation(ctx, indices, r, a.GetData().Slot)
 		} else if err := s.cfg.AttPool.SaveBlockAttestation(a); err != nil {
 			return err
 		}

beacon-chain/blockchain/receive_attestation.go

@@ -177,7 +177,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

@@ -39,18 +39,30 @@ var epochsSinceFinalityExpandCache = primitives.Epoch(4)
 // BlockReceiver interface defines the methods of chain service for receiving and processing new blocks.
 type BlockReceiver interface {
 	ReceiveBlock(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, avs das.AvailabilityStore) error
+	ReceiveExecutionPayloadEnvelope(ctx context.Context, env interfaces.ROExecutionPayloadEnvelope, avs das.AvailabilityStore) error
 	ReceiveBlockBatch(ctx context.Context, blocks []blocks.ROBlock, avs das.AvailabilityStore) error
 	HasBlock(ctx context.Context, root [32]byte) bool
 	RecentBlockSlot(root [32]byte) (primitives.Slot, error)
 	BlockBeingSynced([32]byte) bool
 }
 
+// PayloadAttestationReceiver defines methods of the chain service for receiving
+// and processing new payload attestations and payload attestation messages
+type PayloadAttestationReceiver interface {
+	ReceivePayloadAttestationMessage(ctx context.Context, a *ethpb.PayloadAttestationMessage) error
+}
+
 // BlobReceiver interface defines the methods of chain service for receiving new
 // blobs
 type BlobReceiver interface {
 	ReceiveBlob(context.Context, blocks.VerifiedROBlob) error
 }
 
+// ExecutionPayloadReceiver interface defines the methods of chain service for receiving `ROExecutionPayloadEnvelope`.
+type ExecutionPayloadReceiver interface {
+	ReceiveExecutionPayloadEnvelope(ctx context.Context, envelope interfaces.ROExecutionPayloadEnvelope, _ das.AvailabilityStore) error
+}
+
 // SlashingReceiver interface defines the methods of chain service for receiving validated slashing over the wire.
 type SlashingReceiver interface {
 	ReceiveAttesterSlashing(ctx context.Context, slashing ethpb.AttSlashing)