Receive ptc message (#14394)

* Handle incoming ptc attestation messages in the chain package

* fix double import

* remove unused error

beacon-chain/blockchain/BUILD.bazel

@@ -27,6 +27,7 @@ go_library(
         "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",

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/receive_block.go

@@ -46,6 +46,12 @@ type BlockReceiver interface {
 	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 {

beacon-chain/blockchain/receive_payload_attestation_message.go

@@ -0,0 +1,33 @@
+package blockchain
+
+import (
+	"context"
+	"slices"
+
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+)
+
+func (s *Service) ReceivePayloadAttestationMessage(ctx context.Context, a *eth.PayloadAttestationMessage) error {
+	if err := helpers.ValidateNilPayloadAttestationMessage(a); err != nil {
+		return err
+	}
+	root := [32]byte(a.Data.BeaconBlockRoot)
+	st, err := s.HeadState(ctx)
+	if err != nil {
+		return err
+	}
+	ptc, err := helpers.GetPayloadTimelinessCommittee(ctx, st, a.Data.Slot)
+	if err != nil {
+		return err
+	}
+	idx := slices.Index(ptc, a.ValidatorIndex)
+	if idx == -1 {
+		return errInvalidValidatorIndex
+	}
+	if s.cfg.PayloadAttestationCache.Seen(root, uint64(primitives.ValidatorIndex(idx))) {
+		return nil
+	}
+	return s.cfg.PayloadAttestationCache.Add(a, uint64(idx))
+}

beacon-chain/blockchain/testing/mock.go

@@ -694,3 +694,8 @@ func (c *ChainService) TargetRootForEpoch(_ [32]byte, _ primitives.Epoch) ([32]b
 func (c *ChainService) HashInForkchoice([32]byte) bool {
 	return false
 }
+
+// ReceivePayloadAttestationMessage mocks the same method in the chain service
+func (c *ChainService) ReceivePayloadAttestationMessage(_ context.Context, _ *ethpb.PayloadAttestationMessage) error {
+	return nil
+}

beacon-chain/sync/payload_attestations.go

@@ -2,23 +2,19 @@ package sync
 
 import (
 	"context"
-	"slices"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	payloadattestation "github.com/prysmaticlabs/prysm/v5/consensus-types/epbs/payload-attestation"
-	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"google.golang.org/protobuf/proto"
 )
 
 var (
-	errInvalidValidatorIndex         = errors.New("invalid validator index")
 	errAlreadySeenPayloadAttestation = errors.New("payload attestation already seen for validator index")
 )
 
@@ -91,25 +87,5 @@ func (s *Service) payloadAttestationSubscriber(ctx context.Context, msg proto.Me
 	if !ok {
 		return errWrongMessage
 	}
-	if err := helpers.ValidateNilPayloadAttestationMessage(a); err != nil {
-		return err
-	}
-	root := [32]byte(a.Data.BeaconBlockRoot)
-	st, err := s.cfg.chain.HeadState(ctx)
-	if err != nil {
-		return err
-	}
-	ptc, err := helpers.GetPayloadTimelinessCommittee(ctx, st, a.Data.Slot)
-	if err != nil {
-		return err
-	}
-	idx := slices.Index(ptc, a.ValidatorIndex)
-	if idx == -1 {
-		return errInvalidValidatorIndex
-	}
-	if s.payloadAttestationCache.Seen(root, uint64(primitives.ValidatorIndex(idx))) {
-		return nil
-	}
-
-	return s.payloadAttestationCache.Add(a, uint64(idx))
+	return s.cfg.chain.ReceivePayloadAttestationMessage(ctx, a)
 }

beacon-chain/sync/service.go

@@ -103,6 +103,7 @@ type config struct {
 // This defines the interface for interacting with block chain service
 type blockchainService interface {
 	blockchain.BlockReceiver
+	blockchain.PayloadAttestationReceiver
 	blockchain.BlobReceiver
 	blockchain.HeadFetcher
 	blockchain.FinalizationFetcher