Prysm rpc: Submit payload attestation data (#14381)

beacon-chain/blockchain/receive_payload_attestation_message.go

@@ -14,7 +14,7 @@ func (s *Service) ReceivePayloadAttestationMessage(ctx context.Context, a *eth.P
 		return err
 	}
 	root := [32]byte(a.Data.BeaconBlockRoot)
-	st, err := s.HeadState(ctx)
+	st, err := s.HeadStateReadOnly(ctx)
 	if err != nil {
 		return err
 	}

beacon-chain/blockchain/testing/mock.go

@@ -37,46 +37,47 @@ var ErrNilState = errors.New("nil state")
 
 // ChainService defines the mock interface for testing
 type ChainService struct {
-	NotFinalized                bool
-	Optimistic                  bool
-	ValidAttestation            bool
-	ValidatorsRoot              [32]byte
-	PublicKey                   [fieldparams.BLSPubkeyLength]byte
-	FinalizedCheckPoint         *ethpb.Checkpoint
-	CurrentJustifiedCheckPoint  *ethpb.Checkpoint
-	PreviousJustifiedCheckPoint *ethpb.Checkpoint
-	Slot                        *primitives.Slot // Pointer because 0 is a useful value, so checking against it can be incorrect.
-	Balance                     *precompute.Balance
-	CanonicalRoots              map[[32]byte]bool
-	Fork                        *ethpb.Fork
-	ETH1Data                    *ethpb.Eth1Data
-	InitSyncBlockRoots          map[[32]byte]bool
-	DB                          db.Database
-	State                       state.BeaconState
-	Block                       interfaces.ReadOnlySignedBeaconBlock
-	ExecutionPayloadEnvelope    interfaces.ROExecutionPayloadEnvelope
-	VerifyBlkDescendantErr      error
-	stateNotifier               statefeed.Notifier
-	BlocksReceived              []interfaces.ReadOnlySignedBeaconBlock
-	SyncCommitteeIndices        []primitives.CommitteeIndex
-	blockNotifier               blockfeed.Notifier
-	opNotifier                  opfeed.Notifier
-	Root                        []byte
-	SyncCommitteeDomain         []byte
-	SyncSelectionProofDomain    []byte
-	SyncContributionProofDomain []byte
-	SyncCommitteePubkeys        [][]byte
-	Genesis                     time.Time
-	ForkChoiceStore             forkchoice.ForkChoicer
-	ReceiveBlockMockErr         error
-	ReceiveEnvelopeMockErr      error
-	OptimisticCheckRootReceived [32]byte
-	FinalizedRoots              map[[32]byte]bool
-	OptimisticRoots             map[[32]byte]bool
-	BlockSlot                   primitives.Slot
-	SyncingRoot                 [32]byte
-	Blobs                       []blocks.VerifiedROBlob
-	TargetRoot                  [32]byte
+	NotFinalized                        bool
+	Optimistic                          bool
+	ValidAttestation                    bool
+	ValidatorsRoot                      [32]byte
+	PublicKey                           [fieldparams.BLSPubkeyLength]byte
+	FinalizedCheckPoint                 *ethpb.Checkpoint
+	CurrentJustifiedCheckPoint          *ethpb.Checkpoint
+	PreviousJustifiedCheckPoint         *ethpb.Checkpoint
+	Slot                                *primitives.Slot // Pointer because 0 is a useful value, so checking against it can be incorrect.
+	Balance                             *precompute.Balance
+	CanonicalRoots                      map[[32]byte]bool
+	Fork                                *ethpb.Fork
+	ETH1Data                            *ethpb.Eth1Data
+	InitSyncBlockRoots                  map[[32]byte]bool
+	DB                                  db.Database
+	State                               state.BeaconState
+	Block                               interfaces.ReadOnlySignedBeaconBlock
+	ExecutionPayloadEnvelope            interfaces.ROExecutionPayloadEnvelope
+	VerifyBlkDescendantErr              error
+	stateNotifier                       statefeed.Notifier
+	BlocksReceived                      []interfaces.ReadOnlySignedBeaconBlock
+	SyncCommitteeIndices                []primitives.CommitteeIndex
+	blockNotifier                       blockfeed.Notifier
+	opNotifier                          opfeed.Notifier
+	Root                                []byte
+	SyncCommitteeDomain                 []byte
+	SyncSelectionProofDomain            []byte
+	SyncContributionProofDomain         []byte
+	SyncCommitteePubkeys                [][]byte
+	Genesis                             time.Time
+	ForkChoiceStore                     forkchoice.ForkChoicer
+	ReceiveBlockMockErr                 error
+	ReceiveEnvelopeMockErr              error
+	OptimisticCheckRootReceived         [32]byte
+	FinalizedRoots                      map[[32]byte]bool
+	OptimisticRoots                     map[[32]byte]bool
+	BlockSlot                           primitives.Slot
+	SyncingRoot                         [32]byte
+	Blobs                               []blocks.VerifiedROBlob
+	TargetRoot                          [32]byte
+	ReceivePayloadAttestationMessageErr error
 }
 
 func (s *ChainService) Ancestor(ctx context.Context, root []byte, slot primitives.Slot) ([]byte, error) {
@@ -642,5 +643,5 @@ func (c *ChainService) HashInForkchoice([32]byte) bool {
 
 // ReceivePayloadAttestationMessage mocks the same method in the chain service
 func (c *ChainService) ReceivePayloadAttestationMessage(_ context.Context, _ *ethpb.PayloadAttestationMessage) error {
-	return nil
+	return c.ReceivePayloadAttestationMessageErr
 }

beacon-chain/node/node.go

@@ -999,6 +999,7 @@ func (b *BeaconNode) registerRPCService(router *mux.Router) error {
 		FinalizationFetcher:           chainService,
 		BlockReceiver:                 chainService,
 		BlobReceiver:                  chainService,
+		PayloadAttestationReceiver:    chainService,
 		AttestationReceiver:           chainService,
 		GenesisTimeFetcher:            chainService,
 		GenesisFetcher:                chainService,

beacon-chain/rpc/prysm/v1alpha1/validator/BUILD.bazel

@@ -201,6 +201,7 @@ go_test(
         "proposer_exits_test.go",
         "proposer_slashings_test.go",
         "proposer_sync_aggregate_test.go",
+        "ptc_attester_test.go",
         "proposer_test.go",
         "server_mainnet_test.go",
         "server_test.go",

beacon-chain/rpc/prysm/v1alpha1/validator/ptc_attester.go

@@ -6,12 +6,27 @@ import (
 	"github.com/golang/protobuf/ptypes/empty"
 	"github.com/pkg/errors"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func (vs *Server) GetPayloadAttestationData(ctx context.Context, req *ethpb.GetPayloadAttestationDataRequest) (*ethpb.PayloadAttestationData, error) {
 	return nil, errors.New("not implemented")
 }
 
-func (vs *Server) SubmitPayloadAttestation(ctx context.Context, in *ethpb.PayloadAttestationMessage) (*empty.Empty, error) {
-	return nil, errors.New("not implemented")
+// SubmitPayloadAttestation broadcasts a payload attestation message to the network and saves the payload attestation to the cache.
+// This handler does not validate the payload attestation message before broadcasting and saving it to the cache.
+// The caller should be responsible for validating the message, as it assumes a trusted relationship between the caller and the server.
+func (vs *Server) SubmitPayloadAttestation(ctx context.Context, a *ethpb.PayloadAttestationMessage) (*empty.Empty, error) {
+	// Broadcast the payload attestation message to the network.
+	if err := vs.P2P.Broadcast(ctx, a); err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not broadcast payload attestation message: %v", err)
+	}
+
+	// Save the payload attestation to the cache.
+	if err := vs.PayloadAttestationReceiver.ReceivePayloadAttestationMessage(ctx, a); err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not save payload attestation to cache: %v", err)
+	}
+
+	return &empty.Empty{}, nil
 }

beacon-chain/rpc/prysm/v1alpha1/validator/ptc_attester_test.go

@@ -0,0 +1,41 @@
+package validator
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
+	mock "github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/testing"
+	p2ptest "github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/testing"
+	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+)
+
+func TestServer_SubmitPayloadAttestation(t *testing.T) {
+	ctx := context.Background()
+	t.Run("Error", func(t *testing.T) {
+		s := &Server{
+			P2P:                        p2ptest.NewTestP2P(t),
+			PayloadAttestationReceiver: &mock.ChainService{ReceivePayloadAttestationMessageErr: errors.New("error")},
+		}
+		_, err := s.SubmitPayloadAttestation(ctx, &ethpb.PayloadAttestationMessage{
+			Data: &ethpb.PayloadAttestationData{
+				Slot: 1,
+			},
+		})
+		require.ErrorContains(t, "error", err)
+	})
+
+	t.Run("Happy case", func(t *testing.T) {
+		s := &Server{
+			P2P:                        p2ptest.NewTestP2P(t),
+			PayloadAttestationReceiver: &mock.ChainService{},
+		}
+		_, err := s.SubmitPayloadAttestation(ctx, &ethpb.PayloadAttestationMessage{
+			Data: &ethpb.PayloadAttestationData{
+				Slot: 1,
+			},
+		})
+		require.NoError(t, err)
+	})
+}

beacon-chain/rpc/prysm/v1alpha1/validator/server.go

@@ -43,42 +43,43 @@ import (
 // and committees in which particular validators need to perform their responsibilities,
 // and more.
 type Server struct {
-	Ctx                    context.Context
-	PayloadIDCache         *cache.PayloadIDCache
-	TrackedValidatorsCache *cache.TrackedValidatorsCache
-	HeadFetcher            blockchain.HeadFetcher
-	ForkFetcher            blockchain.ForkFetcher
-	ForkchoiceFetcher      blockchain.ForkchoiceFetcher
-	GenesisFetcher         blockchain.GenesisFetcher
-	FinalizationFetcher    blockchain.FinalizationFetcher
-	TimeFetcher            blockchain.TimeFetcher
-	BlockFetcher           execution.POWBlockFetcher
-	DepositFetcher         cache.DepositFetcher
-	ChainStartFetcher      execution.ChainStartFetcher
-	Eth1InfoFetcher        execution.ChainInfoFetcher
-	OptimisticModeFetcher  blockchain.OptimisticModeFetcher
-	SyncChecker            sync.Checker
-	StateNotifier          statefeed.Notifier
-	BlockNotifier          blockfeed.Notifier
-	P2P                    p2p.Broadcaster
-	AttPool                attestations.Pool
-	SlashingsPool          slashings.PoolManager
-	ExitPool               voluntaryexits.PoolManager
-	SyncCommitteePool      synccommittee.Pool
-	BlockReceiver          blockchain.BlockReceiver
-	BlobReceiver           blockchain.BlobReceiver
-	MockEth1Votes          bool
-	Eth1BlockFetcher       execution.POWBlockFetcher
-	PendingDepositsFetcher depositsnapshot.PendingDepositsFetcher
-	OperationNotifier      opfeed.Notifier
-	StateGen               stategen.StateManager
-	ReplayerBuilder        stategen.ReplayerBuilder
-	BeaconDB               db.HeadAccessDatabase
-	ExecutionEngineCaller  execution.EngineCaller
-	BlockBuilder           builder.BlockBuilder
-	BLSChangesPool         blstoexec.PoolManager
-	ClockWaiter            startup.ClockWaiter
-	CoreService            *core.Service
+	Ctx                        context.Context
+	PayloadIDCache             *cache.PayloadIDCache
+	TrackedValidatorsCache     *cache.TrackedValidatorsCache
+	HeadFetcher                blockchain.HeadFetcher
+	ForkFetcher                blockchain.ForkFetcher
+	ForkchoiceFetcher          blockchain.ForkchoiceFetcher
+	GenesisFetcher             blockchain.GenesisFetcher
+	FinalizationFetcher        blockchain.FinalizationFetcher
+	TimeFetcher                blockchain.TimeFetcher
+	BlockFetcher               execution.POWBlockFetcher
+	DepositFetcher             cache.DepositFetcher
+	ChainStartFetcher          execution.ChainStartFetcher
+	Eth1InfoFetcher            execution.ChainInfoFetcher
+	OptimisticModeFetcher      blockchain.OptimisticModeFetcher
+	SyncChecker                sync.Checker
+	StateNotifier              statefeed.Notifier
+	BlockNotifier              blockfeed.Notifier
+	P2P                        p2p.Broadcaster
+	AttPool                    attestations.Pool
+	SlashingsPool              slashings.PoolManager
+	ExitPool                   voluntaryexits.PoolManager
+	SyncCommitteePool          synccommittee.Pool
+	BlockReceiver              blockchain.BlockReceiver
+	BlobReceiver               blockchain.BlobReceiver
+	PayloadAttestationReceiver blockchain.PayloadAttestationReceiver
+	MockEth1Votes              bool
+	Eth1BlockFetcher           execution.POWBlockFetcher
+	PendingDepositsFetcher     depositsnapshot.PendingDepositsFetcher
+	OperationNotifier          opfeed.Notifier
+	StateGen                   stategen.StateManager
+	ReplayerBuilder            stategen.ReplayerBuilder
+	BeaconDB                   db.HeadAccessDatabase
+	ExecutionEngineCaller      execution.EngineCaller
+	BlockBuilder               builder.BlockBuilder
+	BLSChangesPool             blstoexec.PoolManager
+	ClockWaiter                startup.ClockWaiter
+	CoreService                *core.Service
 }
 
 // WaitForActivation checks if a validator public key exists in the active validator registry of the current

beacon-chain/rpc/service.go

@@ -107,6 +107,7 @@ type Config struct {
 	AttestationReceiver           blockchain.AttestationReceiver
 	BlockReceiver                 blockchain.BlockReceiver
 	BlobReceiver                  blockchain.BlobReceiver
+	PayloadAttestationReceiver    blockchain.PayloadAttestationReceiver
 	ExecutionChainService         execution.Chain
 	ChainStartFetcher             execution.ChainStartFetcher
 	ExecutionChainInfoFetcher     execution.ChainInfoFetcher
@@ -230,42 +231,43 @@ func NewService(ctx context.Context, cfg *Config) *Service {
 		OptimisticModeFetcher: s.cfg.OptimisticModeFetcher,
 	}
 	validatorServer := &validatorv1alpha1.Server{
-		Ctx:                    s.ctx,
-		AttPool:                s.cfg.AttestationsPool,
-		ExitPool:               s.cfg.ExitPool,
-		HeadFetcher:            s.cfg.HeadFetcher,
-		ForkFetcher:            s.cfg.ForkFetcher,
-		ForkchoiceFetcher:      s.cfg.ForkchoiceFetcher,
-		GenesisFetcher:         s.cfg.GenesisFetcher,
-		FinalizationFetcher:    s.cfg.FinalizationFetcher,
-		TimeFetcher:            s.cfg.GenesisTimeFetcher,
-		BlockFetcher:           s.cfg.ExecutionChainService,
-		DepositFetcher:         s.cfg.DepositFetcher,
-		ChainStartFetcher:      s.cfg.ChainStartFetcher,
-		Eth1InfoFetcher:        s.cfg.ExecutionChainService,
-		OptimisticModeFetcher:  s.cfg.OptimisticModeFetcher,
-		SyncChecker:            s.cfg.SyncService,
-		StateNotifier:          s.cfg.StateNotifier,
-		BlockNotifier:          s.cfg.BlockNotifier,
-		OperationNotifier:      s.cfg.OperationNotifier,
-		P2P:                    s.cfg.Broadcaster,
-		BlockReceiver:          s.cfg.BlockReceiver,
-		BlobReceiver:           s.cfg.BlobReceiver,
-		MockEth1Votes:          s.cfg.MockEth1Votes,
-		Eth1BlockFetcher:       s.cfg.ExecutionChainService,
-		PendingDepositsFetcher: s.cfg.PendingDepositFetcher,
-		SlashingsPool:          s.cfg.SlashingsPool,
-		StateGen:               s.cfg.StateGen,
-		SyncCommitteePool:      s.cfg.SyncCommitteeObjectPool,
-		ReplayerBuilder:        ch,
-		ExecutionEngineCaller:  s.cfg.ExecutionEngineCaller,
-		BeaconDB:               s.cfg.BeaconDB,
-		BlockBuilder:           s.cfg.BlockBuilder,
-		BLSChangesPool:         s.cfg.BLSChangesPool,
-		ClockWaiter:            s.cfg.ClockWaiter,
-		CoreService:            coreService,
-		TrackedValidatorsCache: s.cfg.TrackedValidatorsCache,
-		PayloadIDCache:         s.cfg.PayloadIDCache,
+		Ctx:                        s.ctx,
+		AttPool:                    s.cfg.AttestationsPool,
+		ExitPool:                   s.cfg.ExitPool,
+		HeadFetcher:                s.cfg.HeadFetcher,
+		ForkFetcher:                s.cfg.ForkFetcher,
+		ForkchoiceFetcher:          s.cfg.ForkchoiceFetcher,
+		GenesisFetcher:             s.cfg.GenesisFetcher,
+		FinalizationFetcher:        s.cfg.FinalizationFetcher,
+		TimeFetcher:                s.cfg.GenesisTimeFetcher,
+		BlockFetcher:               s.cfg.ExecutionChainService,
+		DepositFetcher:             s.cfg.DepositFetcher,
+		ChainStartFetcher:          s.cfg.ChainStartFetcher,
+		Eth1InfoFetcher:            s.cfg.ExecutionChainService,
+		OptimisticModeFetcher:      s.cfg.OptimisticModeFetcher,
+		SyncChecker:                s.cfg.SyncService,
+		StateNotifier:              s.cfg.StateNotifier,
+		BlockNotifier:              s.cfg.BlockNotifier,
+		OperationNotifier:          s.cfg.OperationNotifier,
+		P2P:                        s.cfg.Broadcaster,
+		BlockReceiver:              s.cfg.BlockReceiver,
+		BlobReceiver:               s.cfg.BlobReceiver,
+		PayloadAttestationReceiver: s.cfg.PayloadAttestationReceiver,
+		MockEth1Votes:              s.cfg.MockEth1Votes,
+		Eth1BlockFetcher:           s.cfg.ExecutionChainService,
+		PendingDepositsFetcher:     s.cfg.PendingDepositFetcher,
+		SlashingsPool:              s.cfg.SlashingsPool,
+		StateGen:                   s.cfg.StateGen,
+		SyncCommitteePool:          s.cfg.SyncCommitteeObjectPool,
+		ReplayerBuilder:            ch,
+		ExecutionEngineCaller:      s.cfg.ExecutionEngineCaller,
+		BeaconDB:                   s.cfg.BeaconDB,
+		BlockBuilder:               s.cfg.BlockBuilder,
+		BLSChangesPool:             s.cfg.BLSChangesPool,
+		ClockWaiter:                s.cfg.ClockWaiter,
+		CoreService:                coreService,
+		TrackedValidatorsCache:     s.cfg.TrackedValidatorsCache,
+		PayloadIDCache:             s.cfg.PayloadIDCache,
 	}
 	s.validatorServer = validatorServer
 	nodeServer := &nodev1alpha1.Server{