Add gossip for payload attestation

beacon-chain/blockchain/BUILD.bazel

@@ -27,6 +27,7 @@ go_library(
         "receive_blob.go",
         "receive_block.go",
         "receive_data_column.go",
+        "receive_payload_attestation_message.go",
         "service.go",
         "setup_forkchoice.go",
         "tracked_proposer.go",

beacon-chain/blockchain/receive_payload_attestation_message.go

@@ -0,0 +1,19 @@
+package blockchain
+
+import (
+	"context"
+
+	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+)
+
+// PayloadAttestationReceiver interface defines the methods of chain service for receiving
+// validated payload attestation messages.
+type PayloadAttestationReceiver interface {
+	ReceivePayloadAttestationMessage(context.Context, *ethpb.PayloadAttestationMessage) error
+}
+
+// ReceivePayloadAttestationMessage accepts a payload attestation message.
+// TODO: implement actual processing; for now this is a stub for sync integration.
+func (s *Service) ReceivePayloadAttestationMessage(ctx context.Context, a *ethpb.PayloadAttestationMessage) error {
+	return nil
+}

beacon-chain/blockchain/testing/mock.go

@@ -757,6 +757,11 @@ func (c *ChainService) ReceiveDataColumns(dcs []blocks.VerifiedRODataColumn) err
 	return nil
 }
 
+// ReceivePayloadAttestationMessage implements the same method in the chain service.
+func (c *ChainService) ReceivePayloadAttestationMessage(_ context.Context, _ *ethpb.PayloadAttestationMessage) error {
+	return nil
+}
+
 // DependentRootForEpoch mocks the same method in the chain service
 func (c *ChainService) DependentRootForEpoch(_ [32]byte, _ primitives.Epoch) ([32]byte, error) {
 	return c.TargetRoot, nil

beacon-chain/cache/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "error.go",
         "interfaces.go",
         "log.go",
+        "payload_attestation.go",
         "payload_id.go",
         "proposer_indices.go",
         "proposer_indices_disabled.go",  # keep

beacon-chain/cache/payload_attestation.go

@@ -0,0 +1,62 @@
+package cache
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
+	eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+)
+
+var errNilPayloadAttestationMessage = errors.New("nil Payload Attestation Message")
+
+// PayloadAttestationCache tracks seen payload attestation messages for a single block root.
+type PayloadAttestationCache struct {
+	root [32]byte
+	seen map[uint64]struct{}
+	sync.Mutex
+}
+
+// Seen returns true if a vote for the given beacon block root has already been
+// processed for this validator index.
+func (p *PayloadAttestationCache) Seen(root [32]byte, idx uint64) bool {
+	p.Lock()
+	defer p.Unlock()
+	if p.root != root {
+		return false
+	}
+	if p.seen == nil {
+		return false
+	}
+	_, ok := p.seen[idx]
+	return ok
+}
+
+// Add marks the given payload attestation message as seen for this validator index.
+// This function assumes that the message has already been validated.
+func (p *PayloadAttestationCache) Add(att *eth.PayloadAttestationMessage, idx uint64) error {
+	if att == nil || att.Data == nil || len(att.Data.BeaconBlockRoot) == 0 {
+		return errNilPayloadAttestationMessage
+	}
+	root := bytesutil.ToBytes32(att.Data.BeaconBlockRoot)
+
+	p.Lock()
+	defer p.Unlock()
+	if p.root != root {
+		p.root = root
+		p.seen = make(map[uint64]struct{})
+	}
+	if p.seen == nil {
+		p.seen = make(map[uint64]struct{})
+	}
+	p.seen[idx] = struct{}{}
+	return nil
+}
+
+// Clear clears the internal cache.
+func (p *PayloadAttestationCache) Clear() {
+	p.Lock()
+	defer p.Unlock()
+	p.root = [32]byte{}
+	p.seen = nil
+}

beacon-chain/core/gloas/payload_attestation.go

@@ -156,6 +156,11 @@ func payloadCommittee(ctx context.Context, st state.ReadOnlyBeaconState, slot pr
 	return selected, nil
 }
 
+// PayloadTimelinessCommittee returns the payload timeliness committee for a given slot.
+func PayloadTimelinessCommittee(ctx context.Context, st state.ReadOnlyBeaconState, slot primitives.Slot) ([]primitives.ValidatorIndex, error) {
+	return payloadCommittee(ctx, st, slot)
+}
+
 // ptcSeed computes the seed for the payload timeliness committee.
 func ptcSeed(st state.ReadOnlyBeaconState, epoch primitives.Epoch, slot primitives.Slot) ([32]byte, error) {
 	seed, err := helpers.Seed(st, epoch, params.BeaconConfig().DomainPTCAttester)

beacon-chain/p2p/gossip_topic_mappings.go

@@ -25,6 +25,7 @@ var gossipTopicMappings = map[string]func() proto.Message{
 	LightClientOptimisticUpdateTopicFormat:    func() proto.Message { return &ethpb.LightClientOptimisticUpdateAltair{} },
 	LightClientFinalityUpdateTopicFormat:      func() proto.Message { return &ethpb.LightClientFinalityUpdateAltair{} },
 	DataColumnSubnetTopicFormat:               func() proto.Message { return &ethpb.DataColumnSidecar{} },
+	PayloadAttestationMessageTopicFormat:      func() proto.Message { return &ethpb.PayloadAttestationMessage{} },
 }
 
 // GossipTopicMappings is a function to return the assigned data type
@@ -144,4 +145,7 @@ func init() {
 
 	// Specially handle Fulu objects.
 	GossipTypeMapping[reflect.TypeFor[*ethpb.SignedBeaconBlockFulu]()] = BlockSubnetTopicFormat
+
+	// Payload attestation messages.
+	GossipTypeMapping[reflect.TypeFor[*ethpb.PayloadAttestationMessage]()] = PayloadAttestationMessageTopicFormat
 }

beacon-chain/p2p/topics.go

@@ -46,6 +46,8 @@ const (
 	GossipLightClientOptimisticUpdateMessage = "light_client_optimistic_update"
 	// GossipDataColumnSidecarMessage is the name for the data column sidecar message type.
 	GossipDataColumnSidecarMessage = "data_column_sidecar"
+	// GossipPayloadAttestationMessage is the name for the payload attestation message type.
+	GossipPayloadAttestationMessage = "payload_attestation_message"
 
 	// Topic Formats
 	//
@@ -75,6 +77,8 @@ const (
 	LightClientOptimisticUpdateTopicFormat = GossipProtocolAndDigest + GossipLightClientOptimisticUpdateMessage
 	// DataColumnSubnetTopicFormat is the topic format for the data column subnet.
 	DataColumnSubnetTopicFormat = GossipProtocolAndDigest + GossipDataColumnSidecarMessage + "_%d"
+	// PayloadAttestationMessageTopicFormat is the topic format for payload attestation messages.
+	PayloadAttestationMessageTopicFormat = GossipProtocolAndDigest + GossipPayloadAttestationMessage
 )
 
 // topic is a struct representing a single gossipsub topic.
@@ -148,6 +152,7 @@ func (s *Service) allTopics() []topic {
 	empty := [4]byte{0, 0, 0, 0} // empty digest for templates, replaced by real digests in per-fork copies.
 	templates := []topic{
 		newTopic(genesis, future, empty, GossipBlockMessage),
+		newTopic(genesis, future, empty, GossipPayloadAttestationMessage),
 		newTopic(genesis, future, empty, GossipAggregateAndProofMessage),
 		newTopic(genesis, future, empty, GossipExitMessage),
 		newTopic(genesis, future, empty, GossipProposerSlashingMessage),

beacon-chain/sync/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "metrics.go",
         "once.go",
         "options.go",
+        "payload_attestations.go",
         "pending_attestations_queue.go",
         "pending_blocks_queue.go",
         "rate_limiter.go",
@@ -113,6 +114,7 @@ go_library(
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//consensus-types/epbs/payload-attestation:go_default_library",
         "//consensus-types/interfaces:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//consensus-types/wrapper:go_default_library",
@@ -177,6 +179,7 @@ go_test(
         "fork_watcher_test.go",
         "kzg_batch_verifier_test.go",
         "once_test.go",
+        "payload_attestations_test.go",
         "pending_attestations_queue_bucket_test.go",
         "pending_attestations_queue_test.go",
         "pending_blocks_queue_test.go",
@@ -263,6 +266,7 @@ go_test(
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//consensus-types/epbs/payload-attestation:go_default_library",
         "//consensus-types/interfaces:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//consensus-types/wrapper:go_default_library",

beacon-chain/sync/options.go

@@ -207,6 +207,13 @@ func WithTrackedValidatorsCache(c *cache.TrackedValidatorsCache) Option {
 	}
 }
 
+func WithPayloadAttestationCache(r *cache.PayloadAttestationCache) Option {
+	return func(s *Service) error {
+		s.payloadAttestationCache = r
+		return nil
+	}
+}
+
 // WithSlasherEnabled configures the sync package to support slashing detection.
 func WithSlasherEnabled(enabled bool) Option {
 	return func(s *Service) error {

beacon-chain/sync/payload_attestations.go

@@ -0,0 +1,94 @@
+package sync
+
+import (
+	"context"
+
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+	"github.com/OffchainLabs/prysm/v7/monitoring/tracing"
+	"github.com/OffchainLabs/prysm/v7/monitoring/tracing/trace"
+	eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/pkg/errors"
+	"google.golang.org/protobuf/proto"
+)
+
+var (
+	errAlreadySeenPayloadAttestation = errors.New("payload attestation already seen for validator index")
+)
+
+func (s *Service) validatePayloadAttestation(ctx context.Context, pid peer.ID, msg *pubsub.Message) (pubsub.ValidationResult, error) {
+	if pid == s.cfg.p2p.PeerID() {
+		return pubsub.ValidationAccept, nil
+	}
+	if s.cfg.initialSync.Syncing() {
+		return pubsub.ValidationIgnore, nil
+	}
+	ctx, span := trace.StartSpan(ctx, "sync.validatePayloadAttestation")
+	defer span.End()
+	if msg.Topic == nil {
+		return pubsub.ValidationReject, p2p.ErrInvalidTopic
+	}
+	m, err := s.decodePubsubMessage(msg)
+	if err != nil {
+		tracing.AnnotateError(span, err)
+		return pubsub.ValidationReject, err
+	}
+	att, ok := m.(*eth.PayloadAttestationMessage)
+	if !ok {
+		return pubsub.ValidationReject, errWrongMessage
+	}
+	pa, err := payloadattestation.NewReadOnly(att)
+	if err != nil {
+		log.WithError(err).Error("Failed to create read only payload attestation")
+		return pubsub.ValidationIgnore, err
+	}
+	v := s.newPayloadAttestationVerifier(pa, verification.GossipPayloadAttestationMessageRequirements)
+
+	if err := v.VerifyCurrentSlot(); err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+
+	if err := v.VerifyPayloadStatus(); err != nil {
+		return pubsub.ValidationReject, err
+	}
+
+	if err := v.VerifyBlockRootSeen(s.hasBadBlock); err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+
+	if err := v.VerifyBlockRootValid(s.hasBadBlock); err != nil {
+		return pubsub.ValidationReject, err
+	}
+
+	st, err := s.cfg.chain.HeadState(ctx)
+	if err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+
+	if err := v.VerifyValidatorInPTC(ctx, st); err != nil {
+		return pubsub.ValidationReject, err
+	}
+
+	if err := v.VerifySignature(st); err != nil {
+		return pubsub.ValidationReject, err
+	}
+
+	if s.payloadAttestationCache.Seen(pa.BeaconBlockRoot(), uint64(pa.ValidatorIndex())) {
+		return pubsub.ValidationIgnore, errAlreadySeenPayloadAttestation
+	}
+
+	msg.ValidatorData = att
+
+	return pubsub.ValidationAccept, nil
+}
+
+func (s *Service) payloadAttestationSubscriber(ctx context.Context, msg proto.Message) error {
+	a, ok := msg.(*eth.PayloadAttestationMessage)
+	if !ok {
+		return errWrongMessage
+	}
+	return s.cfg.chain.ReceivePayloadAttestationMessage(ctx, a)
+}

beacon-chain/sync/payload_attestations_test.go

@@ -0,0 +1,172 @@
+package sync
+
+import (
+	"bytes"
+	"context"
+	"reflect"
+	"testing"
+	"time"
+
+	mock "github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/testing"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p"
+	p2ptest "github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/testing"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/startup"
+	mockSync "github.com/OffchainLabs/prysm/v7/beacon-chain/sync/initial-sync/testing"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
+	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
+	"github.com/OffchainLabs/prysm/v7/config/params"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+	"github.com/OffchainLabs/prysm/v7/testing/require"
+	"github.com/OffchainLabs/prysm/v7/testing/util"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	pb "github.com/libp2p/go-libp2p-pubsub/pb"
+	"github.com/pkg/errors"
+)
+
+func TestValidatePayloadAttestationMessage_IncorrectTopic(t *testing.T) {
+	ctx := context.Background()
+	p := p2ptest.NewTestP2P(t)
+	chainService := &mock.ChainService{Genesis: time.Unix(time.Now().Unix()-int64(params.BeaconConfig().SecondsPerSlot), 0)}
+	s := &Service{
+		payloadAttestationCache: &cache.PayloadAttestationCache{},
+		cfg:                     &config{chain: chainService, p2p: p, initialSync: &mockSync.Sync{}, clock: startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot)}}
+
+	msg := util.HydratePayloadAttestation(&ethpb.PayloadAttestation{}) // Using payload attestation for message should fail.
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, msg)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeFor[*ethpb.PayloadAttestation]()]
+	digest, err := s.currentForkDigest()
+	require.NoError(t, err)
+	topic = s.addDigestToTopic(topic, digest)
+
+	result, err := s.validatePayloadAttestation(ctx, "", &pubsub.Message{
+		Message: &pb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		}})
+	require.ErrorContains(t, "extraction failed for topic", err)
+	require.Equal(t, result, pubsub.ValidationReject)
+}
+
+func TestValidatePayloadAttestationMessage_ErrorPathsWithMock(t *testing.T) {
+	tests := []struct {
+		error    error
+		verifier verification.NewPayloadAttestationMsgVerifier
+		result   pubsub.ValidationResult
+	}{
+		{
+			error: errors.New("incorrect slot"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrIncorrectPayloadAttSlot: errors.New("incorrect slot")}
+			},
+			result: pubsub.ValidationIgnore,
+		},
+		{
+			error: errors.New("incorrect status"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrIncorrectPayloadAttStatus: errors.New("incorrect status")}
+			},
+			result: pubsub.ValidationReject,
+		},
+		{
+			error: errors.New("block root seen"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrPayloadAttBlockRootNotSeen: errors.New("block root seen")}
+			},
+			result: pubsub.ValidationIgnore,
+		},
+		{
+			error: errors.New("block root invalid"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrPayloadAttBlockRootInvalid: errors.New("block root invalid")}
+			},
+			result: pubsub.ValidationReject,
+		},
+		{
+			error: errors.New("validator not in PTC"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrIncorrectPayloadAttValidator: errors.New("validator not in PTC")}
+			},
+			result: pubsub.ValidationReject,
+		},
+		{
+			error: errors.New("incorrect signature"),
+			verifier: func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+				return &verification.MockPayloadAttestation{ErrInvalidMessageSignature: errors.New("incorrect signature")}
+			},
+			result: pubsub.ValidationReject,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.error.Error(), func(t *testing.T) {
+			ctx := context.Background()
+			p := p2ptest.NewTestP2P(t)
+			chainService := &mock.ChainService{Genesis: time.Unix(time.Now().Unix()-int64(params.BeaconConfig().SecondsPerSlot), 0)}
+			s := &Service{
+				payloadAttestationCache: &cache.PayloadAttestationCache{},
+				cfg:                     &config{chain: chainService, p2p: p, initialSync: &mockSync.Sync{}, clock: startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot)}}
+			s.newPayloadAttestationVerifier = tt.verifier
+
+			msg := newPayloadAttestationMessage()
+			buf := new(bytes.Buffer)
+			_, err := p.Encoding().EncodeGossip(buf, msg)
+			require.NoError(t, err)
+
+			topic := p2p.GossipTypeMapping[reflect.TypeFor[*ethpb.PayloadAttestationMessage]()]
+			digest, err := s.currentForkDigest()
+			require.NoError(t, err)
+			topic = s.addDigestToTopic(topic, digest)
+
+			result, err := s.validatePayloadAttestation(ctx, "", &pubsub.Message{
+				Message: &pb.Message{
+					Data:  buf.Bytes(),
+					Topic: &topic,
+				}})
+
+			require.ErrorContains(t, tt.error.Error(), err)
+			require.Equal(t, result, tt.result)
+		})
+	}
+}
+
+func TestValidatePayloadAttestationMessage_Accept(t *testing.T) {
+	ctx := context.Background()
+	p := p2ptest.NewTestP2P(t)
+	chainService := &mock.ChainService{Genesis: time.Unix(time.Now().Unix()-int64(params.BeaconConfig().SecondsPerSlot), 0)}
+	s := &Service{
+		payloadAttestationCache: &cache.PayloadAttestationCache{},
+		cfg:                     &config{chain: chainService, p2p: p, initialSync: &mockSync.Sync{}, clock: startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot)}}
+	s.newPayloadAttestationVerifier = func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+		return &verification.MockPayloadAttestation{}
+	}
+
+	msg := newPayloadAttestationMessage()
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, msg)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeFor[*ethpb.PayloadAttestationMessage]()]
+	digest, err := s.currentForkDigest()
+	require.NoError(t, err)
+	topic = s.addDigestToTopic(topic, digest)
+
+	result, err := s.validatePayloadAttestation(ctx, "", &pubsub.Message{
+		Message: &pb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		}})
+	require.NoError(t, err)
+	require.Equal(t, result, pubsub.ValidationAccept)
+}
+
+func newPayloadAttestationMessage() *ethpb.PayloadAttestationMessage {
+	return &ethpb.PayloadAttestationMessage{
+		ValidatorIndex: 0,
+		Data:           util.HydratePayloadAttestationData(&ethpb.PayloadAttestationData{Slot: 1}),
+		Signature:      make([]byte, fieldparams.BLSSignatureLength),
+	}
+}

beacon-chain/sync/service.go

@@ -37,6 +37,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	leakybucket "github.com/OffchainLabs/prysm/v7/container/leaky-bucket"
@@ -121,6 +122,7 @@ type blockchainService interface {
 	blockchain.FinalizationFetcher
 	blockchain.ForkFetcher
 	blockchain.AttestationReceiver
+	blockchain.PayloadAttestationReceiver
 	blockchain.TimeFetcher
 	blockchain.GenesisFetcher
 	blockchain.CanonicalFetcher
@@ -173,6 +175,7 @@ type Service struct {
 	verifierWaiter                   *verification.InitializerWaiter
 	newBlobVerifier                  verification.NewBlobVerifier
 	newColumnsVerifier               verification.NewDataColumnsVerifier
+	newPayloadAttestationVerifier    verification.NewPayloadAttestationMsgVerifier
 	columnSidecarsExecSingleFlight   singleflight.Group
 	reconstructionSingleFlight       singleflight.Group
 	availableBlocker                 coverage.AvailableBlocker
@@ -182,6 +185,7 @@ type Service struct {
 	slasherEnabled                   bool
 	lcStore                          *lightClient.Store
 	dataColumnLogCh                  chan dataColumnLogEntry
+	payloadAttestationCache          *cache.PayloadAttestationCache
 	digestActions                    perDigestSet
 	subscriptionSpawner              func(func()) // see Service.spawn for details
 }
@@ -190,15 +194,16 @@ type Service struct {
 func NewService(ctx context.Context, opts ...Option) *Service {
 	ctx, cancel := context.WithCancel(ctx)
 	r := &Service{
-		ctx:                   ctx,
-		cancel:                cancel,
-		chainStarted:          abool.New(),
-		cfg:                   &config{clock: startup.NewClock(time.Unix(0, 0), [32]byte{})},
-		slotToPendingBlocks:   gcache.New(pendingBlockExpTime /* exp time */, 0 /* disable janitor */),
-		seenPendingBlocks:     make(map[[32]byte]bool),
-		blkRootToPendingAtts:  make(map[[32]byte][]any),
-		dataColumnLogCh:       make(chan dataColumnLogEntry, 1000),
-		reconstructionRandGen: rand.NewGenerator(),
+		ctx:                     ctx,
+		cancel:                  cancel,
+		chainStarted:            abool.New(),
+		cfg:                     &config{clock: startup.NewClock(time.Unix(0, 0), [32]byte{})},
+		slotToPendingBlocks:     gcache.New(pendingBlockExpTime /* exp time */, 0 /* disable janitor */),
+		seenPendingBlocks:       make(map[[32]byte]bool),
+		blkRootToPendingAtts:    make(map[[32]byte][]any),
+		dataColumnLogCh:         make(chan dataColumnLogEntry, 1000),
+		reconstructionRandGen:   rand.NewGenerator(),
+		payloadAttestationCache: &cache.PayloadAttestationCache{},
 	}
 
 	for _, opt := range opts {
@@ -250,6 +255,12 @@ func newDataColumnsVerifierFromInitializer(ini *verification.Initializer) verifi
 	}
 }
 
+func newPayloadAttestationMessageFromInitializer(ini *verification.Initializer) verification.NewPayloadAttestationMsgVerifier {
+	return func(pa payloadattestation.ROMessage, reqs []verification.Requirement) verification.PayloadAttestationMsgVerifier {
+		return ini.NewPayloadAttestationMsgVerifier(pa, reqs)
+	}
+}
+
 // Start the regular sync service.
 func (s *Service) Start() {
 	v, err := s.verifierWaiter.WaitForInitializer(s.ctx)
@@ -259,6 +270,7 @@ func (s *Service) Start() {
 	}
 	s.newBlobVerifier = newBlobVerifierFromInitializer(v)
 	s.newColumnsVerifier = newDataColumnsVerifierFromInitializer(v)
+	s.newPayloadAttestationVerifier = newPayloadAttestationMessageFromInitializer(v)
 
 	go s.verifierRoutine()
 	go s.startDiscoveryAndSubscriptions()

beacon-chain/verification/BUILD.bazel

@@ -7,20 +7,25 @@ go_library(
         "blob.go",
         "cache.go",
         "data_column.go",
+        "epbs.go",
         "error.go",
         "fake.go",
         "filesystem.go",
         "initializer.go",
+        "initializer_epbs.go",
         "interface.go",
         "log.go",
         "metrics.go",
         "mock.go",
+        "payload_attestation.go",
+        "payload_attestation_mock.go",
         "result.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/verification",
     visibility = ["//visibility:public"],
     deps = [
         "//beacon-chain/blockchain/kzg:go_default_library",
+        "//beacon-chain/core/gloas:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
@@ -32,6 +37,7 @@ go_library(
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
+        "//consensus-types/epbs/payload-attestation:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//crypto/bls:go_default_library",
         "//encoding/bytesutil:go_default_library",

beacon-chain/verification/epbs.go

@@ -0,0 +1,24 @@
+package verification
+
+import (
+	"context"
+
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+)
+
+// PayloadAttestationMsgVerifier defines the methods implemented by the ROPayloadAttestation.
+type PayloadAttestationMsgVerifier interface {
+	VerifyCurrentSlot() error
+	VerifyPayloadStatus() error
+	VerifyBlockRootSeen(func([32]byte) bool) error
+	VerifyBlockRootValid(func([32]byte) bool) error
+	VerifyValidatorInPTC(context.Context, state.BeaconState) error
+	VerifySignature(state.BeaconState) error
+	VerifiedPayloadAttestation() (payloadattestation.VerifiedROMessage, error)
+	SatisfyRequirement(Requirement)
+}
+
+// NewPayloadAttestationMsgVerifier is a function signature that can be used by code that needs to be
+// able to mock Initializer.NewPayloadAttestationMsgVerifier without complex setup.
+type NewPayloadAttestationMsgVerifier func(pa payloadattestation.ROMessage, reqs []Requirement) PayloadAttestationMsgVerifier

beacon-chain/verification/initializer_epbs.go

@@ -0,0 +1,15 @@
+package verification
+
+import (
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+)
+
+// NewPayloadAttestationMsgVerifier creates a PayloadAttestationMsgVerifier for a single payload attestation message,
+// with the given set of requirements.
+func (ini *Initializer) NewPayloadAttestationMsgVerifier(pa payloadattestation.ROMessage, reqs []Requirement) *PayloadAttMsgVerifier {
+	return &PayloadAttMsgVerifier{
+		sharedResources: ini.shared,
+		results:         newResults(reqs...),
+		pa:              pa,
+	}
+}

beacon-chain/verification/payload_attestation.go

@@ -0,0 +1,218 @@
+package verification
+
+import (
+	"context"
+	"fmt"
+	"slices"
+
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/gloas"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/signing"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
+	"github.com/OffchainLabs/prysm/v7/config/params"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+	"github.com/OffchainLabs/prysm/v7/crypto/bls"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+// RequirementList defines a list of requirements.
+type RequirementList []Requirement
+
+const (
+	RequireCurrentSlot Requirement = iota
+	RequireMessageNotSeen
+	RequireKnownPayloadStatus
+	RequireValidatorInPTC
+	RequireBlockRootSeen
+	RequireBlockRootValid
+	RequireSignatureValid
+)
+
+// PayloadAttGossipRequirements defines the list of requirements for gossip payload attestation messages.
+var PayloadAttGossipRequirements = []Requirement{
+	RequireCurrentSlot,
+	RequireMessageNotSeen,
+	RequireKnownPayloadStatus,
+	RequireValidatorInPTC,
+	RequireBlockRootSeen,
+	RequireBlockRootValid,
+	RequireSignatureValid,
+}
+
+// GossipPayloadAttestationMessageRequirements is a requirement list for gossip payload attestation messages.
+var GossipPayloadAttestationMessageRequirements = RequirementList(PayloadAttGossipRequirements)
+
+var (
+	ErrIncorrectPayloadAttSlot      = errors.New("payload att slot does not match the current slot")
+	ErrIncorrectPayloadAttStatus    = errors.New("unknown payload att status")
+	ErrPayloadAttBlockRootNotSeen   = errors.New("block root not seen")
+	ErrPayloadAttBlockRootInvalid   = errors.New("block root invalid")
+	ErrIncorrectPayloadAttValidator = errors.New("validator not present in payload timeliness committee")
+	ErrInvalidPayloadAttMessage     = errors.New("invalid payload attestation message")
+)
+
+var _ PayloadAttestationMsgVerifier = &PayloadAttMsgVerifier{}
+
+// PayloadAttMsgVerifier is a read-only verifier for payload attestation messages.
+type PayloadAttMsgVerifier struct {
+	*sharedResources
+	results *results
+	pa      payloadattestation.ROMessage
+}
+
+// VerifyCurrentSlot verifies if the current slot matches the expected slot.
+// Represents the following spec verification:
+// [IGNORE] data.slot is the current slot.
+func (v *PayloadAttMsgVerifier) VerifyCurrentSlot() (err error) {
+	defer v.record(RequireCurrentSlot, &err)
+
+	if v.pa.Slot() != v.clock.CurrentSlot() {
+		log.WithFields(logFields(v.pa)).Errorf("does not match current slot %d", v.clock.CurrentSlot())
+		return ErrIncorrectPayloadAttSlot
+	}
+
+	return nil
+}
+
+// VerifyPayloadStatus verifies if the payload status is known.
+// In Gloas, payload status is represented by booleans, so there is no invalid enum to reject.
+func (v *PayloadAttMsgVerifier) VerifyPayloadStatus() (err error) {
+	defer v.record(RequireKnownPayloadStatus, &err)
+	return nil
+}
+
+// VerifyBlockRootSeen verifies if the block root has been seen before.
+// Represents the following spec verification:
+// [IGNORE] The attestation's data.beacon_block_root has been seen (via both gossip and non-gossip sources).
+func (v *PayloadAttMsgVerifier) VerifyBlockRootSeen(parentSeen func([32]byte) bool) (err error) {
+	defer v.record(RequireBlockRootSeen, &err)
+	if parentSeen != nil && parentSeen(v.pa.BeaconBlockRoot()) {
+		return nil
+	}
+	log.WithFields(logFields(v.pa)).Error(ErrPayloadAttBlockRootNotSeen.Error())
+	return ErrPayloadAttBlockRootNotSeen
+}
+
+// VerifyBlockRootValid verifies if the block root is valid.
+// Represents the following spec verification:
+// [REJECT] The beacon block with root data.beacon_block_root passes validation.
+func (v *PayloadAttMsgVerifier) VerifyBlockRootValid(badBlock func([32]byte) bool) (err error) {
+	defer v.record(RequireBlockRootValid, &err)
+
+	if badBlock != nil && badBlock(v.pa.BeaconBlockRoot()) {
+		log.WithFields(logFields(v.pa)).Error(ErrPayloadAttBlockRootInvalid.Error())
+		return ErrPayloadAttBlockRootInvalid
+	}
+
+	return nil
+}
+
+// VerifyValidatorInPTC verifies if the validator is present.
+// Represents the following spec verification:
+// [REJECT] The validator index is within the payload committee in get_ptc(state, data.slot). For the current's slot head state.
+func (v *PayloadAttMsgVerifier) VerifyValidatorInPTC(ctx context.Context, st state.BeaconState) (err error) {
+	defer v.record(RequireValidatorInPTC, &err)
+
+	ptc, err := gloas.PayloadTimelinessCommittee(ctx, st, v.pa.Slot())
+	if err != nil {
+		return err
+	}
+
+	idx := slices.Index(ptc, v.pa.ValidatorIndex())
+	if idx == -1 {
+		log.WithFields(logFields(v.pa)).Error(ErrIncorrectPayloadAttValidator.Error())
+		return ErrIncorrectPayloadAttValidator
+	}
+
+	return nil
+}
+
+// VerifySignature verifies the signature of the payload attestation message.
+// Represents the following spec verification:
+// [REJECT] The signature of payload_attestation_message.signature is valid with respect to the validator index.
+func (v *PayloadAttMsgVerifier) VerifySignature(st state.BeaconState) (err error) {
+	defer v.record(RequireSignatureValid, &err)
+
+	err = validatePayloadAttestationMessageSignature(st, v.pa)
+	if err != nil {
+		if errors.Is(err, signing.ErrSigFailedToVerify) {
+			log.WithFields(logFields(v.pa)).Error("Signature failed to validate")
+		} else {
+			log.WithFields(logFields(v.pa)).WithError(err).Error("Could not validate signature")
+		}
+		return err
+	}
+
+	return nil
+}
+
+// VerifiedPayloadAttestation returns a verified payload attestation message by checking all requirements.
+func (v *PayloadAttMsgVerifier) VerifiedPayloadAttestation() (payloadattestation.VerifiedROMessage, error) {
+	if v.results.allSatisfied() {
+		return payloadattestation.NewVerifiedROMessage(v.pa), nil
+	}
+	return payloadattestation.VerifiedROMessage{}, ErrInvalidPayloadAttMessage
+}
+
+// SatisfyRequirement allows the caller to manually mark a requirement as satisfied.
+func (v *PayloadAttMsgVerifier) SatisfyRequirement(req Requirement) {
+	v.record(req, nil)
+}
+
+// ValidatePayloadAttestationMessageSignature verifies the signature of a payload attestation message.
+func validatePayloadAttestationMessageSignature(st state.BeaconState, payloadAtt payloadattestation.ROMessage) error {
+	val, err := st.ValidatorAtIndex(payloadAtt.ValidatorIndex())
+	if err != nil {
+		return err
+	}
+
+	pub, err := bls.PublicKeyFromBytes(val.PublicKey)
+	if err != nil {
+		return err
+	}
+
+	s := payloadAtt.Signature()
+	sig, err := bls.SignatureFromBytes(s[:])
+	if err != nil {
+		return err
+	}
+
+	currentEpoch := slots.ToEpoch(st.Slot())
+	domain, err := signing.Domain(st.Fork(), currentEpoch, params.BeaconConfig().DomainPTCAttester, st.GenesisValidatorsRoot())
+	if err != nil {
+		return err
+	}
+
+	root, err := payloadAtt.SigningRoot(domain)
+	if err != nil {
+		return err
+	}
+
+	if !sig.Verify(pub, root[:]) {
+		return signing.ErrSigFailedToVerify
+	}
+	return nil
+}
+
+// record records the result of a requirement verification.
+func (v *PayloadAttMsgVerifier) record(req Requirement, err *error) {
+	if err == nil || *err == nil {
+		v.results.record(req, nil)
+		return
+	}
+
+	v.results.record(req, *err)
+}
+
+// logFields returns log fields for a ROMessage instance.
+func logFields(payload payloadattestation.ROMessage) logrus.Fields {
+	return logrus.Fields{
+		"slot":              payload.Slot(),
+		"validatorIndex":    payload.ValidatorIndex(),
+		"signature":         fmt.Sprintf("%#x", payload.Signature()),
+		"beaconBlockRoot":   fmt.Sprintf("%#x", payload.BeaconBlockRoot()),
+		"payloadPresent":    payload.PayloadPresent(),
+		"blobDataAvailable": payload.BlobDataAvailable(),
+	}
+}

beacon-chain/verification/payload_attestation_mock.go

@@ -0,0 +1,51 @@
+package verification
+
+import (
+	"context"
+
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
+	payloadattestation "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation"
+)
+
+type MockPayloadAttestation struct {
+	ErrIncorrectPayloadAttSlot      error
+	ErrIncorrectPayloadAttStatus    error
+	ErrIncorrectPayloadAttValidator error
+	ErrPayloadAttBlockRootNotSeen   error
+	ErrPayloadAttBlockRootInvalid   error
+	ErrInvalidPayloadAttMessage     error
+	ErrInvalidMessageSignature      error
+	ErrUnsatisfiedRequirement       error
+}
+
+var _ PayloadAttestationMsgVerifier = &MockPayloadAttestation{}
+
+func (m *MockPayloadAttestation) VerifyCurrentSlot() error {
+	return m.ErrIncorrectPayloadAttSlot
+}
+
+func (m *MockPayloadAttestation) VerifyPayloadStatus() error {
+	return m.ErrIncorrectPayloadAttStatus
+}
+
+func (m *MockPayloadAttestation) VerifyValidatorInPTC(ctx context.Context, st state.BeaconState) error {
+	return m.ErrIncorrectPayloadAttValidator
+}
+
+func (m *MockPayloadAttestation) VerifyBlockRootSeen(func([32]byte) bool) error {
+	return m.ErrPayloadAttBlockRootNotSeen
+}
+
+func (m *MockPayloadAttestation) VerifyBlockRootValid(func([32]byte) bool) error {
+	return m.ErrPayloadAttBlockRootInvalid
+}
+
+func (m *MockPayloadAttestation) VerifySignature(st state.BeaconState) (err error) {
+	return m.ErrInvalidMessageSignature
+}
+
+func (m *MockPayloadAttestation) VerifiedPayloadAttestation() (payloadattestation.VerifiedROMessage, error) {
+	return payloadattestation.VerifiedROMessage{}, nil
+}
+
+func (m *MockPayloadAttestation) SatisfyRequirement(req Requirement) {}

consensus-types/epbs/payload-attestation/BUILD.bazel

@@ -0,0 +1,15 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["readonly_message.go"],
+    importpath = "github.com/OffchainLabs/prysm/v7/consensus-types/epbs/payload-attestation",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//beacon-chain/core/signing:go_default_library",
+        "//consensus-types/primitives:go_default_library",
+        "//encoding/bytesutil:go_default_library",
+        "//proto/prysm/v1alpha1:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
+    ],
+)

consensus-types/epbs/payload-attestation/readonly_message.go

@@ -0,0 +1,87 @@
+package payloadattestation
+
+import (
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/signing"
+	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
+	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
+	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+	"github.com/pkg/errors"
+)
+
+var (
+	errNilPayloadAttMessage   = errors.New("received nil payload attestation message")
+	errNilPayloadAttData      = errors.New("received nil payload attestation data")
+	errNilPayloadAttSignature = errors.New("received nil payload attestation signature")
+)
+
+// ROMessage represents a read-only payload attestation message.
+type ROMessage struct {
+	m *ethpb.PayloadAttestationMessage
+}
+
+// validatePayloadAtt checks if the given payload attestation message is valid.
+func validatePayloadAtt(m *ethpb.PayloadAttestationMessage) error {
+	if m == nil {
+		return errNilPayloadAttMessage
+	}
+	if m.Data == nil {
+		return errNilPayloadAttData
+	}
+	if len(m.Signature) == 0 {
+		return errNilPayloadAttSignature
+	}
+	return nil
+}
+
+// NewReadOnly creates a new ReadOnly instance after validating the message.
+func NewReadOnly(m *ethpb.PayloadAttestationMessage) (ROMessage, error) {
+	if err := validatePayloadAtt(m); err != nil {
+		return ROMessage{}, err
+	}
+	return ROMessage{m}, nil
+}
+
+// ValidatorIndex returns the validator index from the payload attestation message.
+func (r *ROMessage) ValidatorIndex() primitives.ValidatorIndex {
+	return r.m.ValidatorIndex
+}
+
+// Signature returns the signature from the payload attestation message.
+func (r *ROMessage) Signature() [96]byte {
+	return bytesutil.ToBytes96(r.m.Signature)
+}
+
+// BeaconBlockRoot returns the beacon block root from the payload attestation message.
+func (r *ROMessage) BeaconBlockRoot() [32]byte {
+	return bytesutil.ToBytes32(r.m.Data.BeaconBlockRoot)
+}
+
+// Slot returns the slot from the payload attestation message.
+func (r *ROMessage) Slot() primitives.Slot {
+	return r.m.Data.Slot
+}
+
+// PayloadPresent returns whether the payload was present.
+func (r *ROMessage) PayloadPresent() bool {
+	return r.m.Data.PayloadPresent
+}
+
+// BlobDataAvailable returns whether blob data was available.
+func (r *ROMessage) BlobDataAvailable() bool {
+	return r.m.Data.BlobDataAvailable
+}
+
+// SigningRoot returns the signing root from the payload attestation message.
+func (r *ROMessage) SigningRoot(domain []byte) ([32]byte, error) {
+	return signing.ComputeSigningRoot(r.m.Data, domain)
+}
+
+// VerifiedROMessage represents a verified read-only payload attestation message.
+type VerifiedROMessage struct {
+	ROMessage
+}
+
+// NewVerifiedROMessage creates a new VerifiedROMessage instance after validating the message.
+func NewVerifiedROMessage(r ROMessage) VerifiedROMessage {
+	return VerifiedROMessage{r}
+}