Payload Attestation Sync package changes

beacon-chain/blockchain/options.go

@@ -69,6 +69,14 @@ 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
+	}
+}
+
 // WithPayloadIDCache for payload ID cache.
 func WithPayloadIDCache(c *cache.PayloadIDCache) Option {
 	return func(s *Service) error {

beacon-chain/blockchain/service.go

@@ -76,6 +76,7 @@ type config struct {
 	ChainStartFetcher       execution.ChainStartFetcher
 	BeaconDB                db.HeadAccessDatabase
 	DepositCache            cache.DepositCache
+	PayloadAttestationCache *cache.PayloadAttestationCache
 	PayloadIDCache          *cache.PayloadIDCache
 	TrackedValidatorsCache  *cache.TrackedValidatorsCache
 	AttPool                 attestations.Pool

beacon-chain/cache/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "doc.go",
         "error.go",
         "interfaces.go",
+        "payload_attestation.go",
         "payload_id.go",
         "proposer_indices.go",
         "proposer_indices_disabled.go",  # keep
@@ -42,6 +43,7 @@ go_library(
         "//config/params:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//container/slice:go_default_library",
+        "//crypto/bls:go_default_library",
         "//crypto/hash:go_default_library",
         "//crypto/rand:go_default_library",
         "//encoding/bytesutil:go_default_library",
@@ -70,6 +72,7 @@ go_test(
         "checkpoint_state_test.go",
         "committee_fuzz_test.go",
         "committee_test.go",
+        "payload_attestation_test.go",
         "payload_id_test.go",
         "private_access_test.go",
         "proposer_indices_test.go",
@@ -88,6 +91,7 @@ go_test(
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/primitives:go_default_library",
+        "//crypto/bls:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
         "//testing/assert:go_default_library",

beacon-chain/cache/payload_attestation.go

@@ -0,0 +1,116 @@
+package cache
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/crypto/bls"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+)
+
+var errNilPayloadAttestationMessage = errors.New("nil Payload Attestation Message")
+
+// PayloadAttestationCache keeps a map of all the PTC votes that were seen,
+// already aggregated. The key is the beacon block root.
+type PayloadAttestationCache struct {
+	root         [32]byte
+	attestations [primitives.PAYLOAD_INVALID_STATUS]*eth.PayloadAttestation
+	sync.Mutex
+}
+
+// Seen returns true if a vote for the given Beacon Block Root has already been processed
+// for this Payload Timeliness Committee index. This will return true even if
+// the Payload status differs.
+func (p *PayloadAttestationCache) Seen(root [32]byte, idx uint64) bool {
+	p.Lock()
+	defer p.Unlock()
+	if p.root != root {
+		return false
+	}
+	for _, agg := range p.attestations {
+		if agg == nil {
+			continue
+		}
+		if agg.AggregationBits.BitAt(idx) {
+			return true
+		}
+	}
+	return false
+}
+
+// messageToPayloadAttestation creates a PayloadAttestation with a single
+// aggregated bit from the passed PayloadAttestationMessage
+func messageToPayloadAttestation(att *eth.PayloadAttestationMessage, idx uint64) *eth.PayloadAttestation {
+	bits := primitives.NewPayloadAttestationAggregationBits()
+	bits.SetBitAt(idx, true)
+	data := &eth.PayloadAttestationData{
+		BeaconBlockRoot: bytesutil.SafeCopyBytes(att.Data.BeaconBlockRoot),
+		Slot:            att.Data.Slot,
+		PayloadStatus:   att.Data.PayloadStatus,
+	}
+	return &eth.PayloadAttestation{
+		AggregationBits: bits,
+		Data:            data,
+		Signature:       bytesutil.SafeCopyBytes(att.Signature),
+	}
+}
+
+// aggregateSigFromMessage returns the aggregated signature from a Payload
+// Attestation by adding the passed signature in the PayloadAttestationMessage,
+// no signature validation is performed.
+func aggregateSigFromMessage(aggregated *eth.PayloadAttestation, message *eth.PayloadAttestationMessage) ([]byte, error) {
+	aggSig, err := bls.SignatureFromBytesNoValidation(aggregated.Signature)
+	if err != nil {
+		return nil, err
+	}
+	sig, err := bls.SignatureFromBytesNoValidation(message.Signature)
+	if err != nil {
+		return nil, err
+	}
+	return bls.AggregateSignatures([]bls.Signature{aggSig, sig}).Marshal(), nil
+}
+
+// Add adds a PayloadAttestationMessage to the internal cache of aggregated
+// PayloadAttestations.
+// If the index has already been seen for this attestation status the function does nothing.
+// If the root is not the cached root, the function will clear the previous cache
+// This function assumes that the message has already been validated. In
+// particular that the signature is valid and that the block root corresponds to
+// the given slot in the attestation data.
+func (p *PayloadAttestationCache) Add(att *eth.PayloadAttestationMessage, idx uint64) error {
+	if att == nil || att.Data == nil || att.Data.BeaconBlockRoot == nil {
+		return errNilPayloadAttestationMessage
+	}
+	p.Lock()
+	defer p.Unlock()
+	root := [32]byte(att.Data.BeaconBlockRoot)
+	if p.root != root {
+		p.root = root
+		p.attestations = [primitives.PAYLOAD_INVALID_STATUS]*eth.PayloadAttestation{}
+	}
+	agg := p.attestations[att.Data.PayloadStatus]
+	if agg == nil {
+		p.attestations[att.Data.PayloadStatus] = messageToPayloadAttestation(att, idx)
+		return nil
+	}
+	if agg.AggregationBits.BitAt(idx) {
+		return nil
+	}
+	sig, err := aggregateSigFromMessage(agg, att)
+	if err != nil {
+		return err
+	}
+	agg.Signature = sig
+	agg.AggregationBits.SetBitAt(idx, true)
+	return nil
+}
+
+// Clear clears the internal map
+func (p *PayloadAttestationCache) Clear() {
+	p.Lock()
+	defer p.Unlock()
+	p.root = [32]byte{}
+	p.attestations = [primitives.PAYLOAD_INVALID_STATUS]*eth.PayloadAttestation{}
+}

beacon-chain/cache/payload_attestation_test.go

@@ -0,0 +1,95 @@
+package cache
+
+import (
+	"testing"
+
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/crypto/bls"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+)
+
+func TestPayloadAttestationCache(t *testing.T) {
+	p := &PayloadAttestationCache{}
+
+	//Test Has seen
+	root := [32]byte{'r'}
+	idx := uint64(5)
+	require.Equal(t, false, p.Seen(root, idx))
+
+	// Test Add
+	msg := &eth.PayloadAttestationMessage{
+		Signature: bls.NewAggregateSignature().Marshal(),
+		Data: &eth.PayloadAttestationData{
+			BeaconBlockRoot: root[:],
+			Slot:            1,
+			PayloadStatus:   primitives.PAYLOAD_PRESENT,
+		},
+	}
+
+	// Add new root
+	require.NoError(t, p.Add(msg, idx))
+	require.Equal(t, true, p.Seen(root, idx))
+	require.Equal(t, root, p.root)
+	att := p.attestations[primitives.PAYLOAD_PRESENT]
+	indices := att.AggregationBits.BitIndices()
+	require.DeepEqual(t, []int{int(idx)}, indices)
+	singleSig := bytesutil.SafeCopyBytes(msg.Signature)
+	require.DeepEqual(t, singleSig, att.Signature)
+
+	// Test Seen
+	require.Equal(t, true, p.Seen(root, idx))
+	require.Equal(t, false, p.Seen(root, idx+1))
+
+	// Add another attestation on the same data
+	msg2 := &eth.PayloadAttestationMessage{
+		Signature: bls.NewAggregateSignature().Marshal(),
+		Data:      att.Data,
+	}
+	idx2 := uint64(7)
+	require.NoError(t, p.Add(msg2, idx2))
+	att = p.attestations[primitives.PAYLOAD_PRESENT]
+	indices = att.AggregationBits.BitIndices()
+	require.DeepEqual(t, []int{int(idx), int(idx2)}, indices)
+	require.DeepNotEqual(t, att.Signature, msg.Signature)
+
+	// Try again the same index
+	require.NoError(t, p.Add(msg2, idx2))
+	att2 := p.attestations[primitives.PAYLOAD_PRESENT]
+	indices = att.AggregationBits.BitIndices()
+	require.DeepEqual(t, []int{int(idx), int(idx2)}, indices)
+	require.DeepEqual(t, att, att2)
+
+	// Test Seen
+	require.Equal(t, true, p.Seen(root, idx2))
+	require.Equal(t, false, p.Seen(root, idx2+1))
+
+	// Add another payload status for a different payload status
+	msg3 := &eth.PayloadAttestationMessage{
+		Signature: bls.NewAggregateSignature().Marshal(),
+		Data: &eth.PayloadAttestationData{
+			BeaconBlockRoot: root[:],
+			Slot:            1,
+			PayloadStatus:   primitives.PAYLOAD_WITHHELD,
+		},
+	}
+	idx3 := uint64(17)
+
+	require.NoError(t, p.Add(msg3, idx3))
+	att3 := p.attestations[primitives.PAYLOAD_WITHHELD]
+	indices3 := att3.AggregationBits.BitIndices()
+	require.DeepEqual(t, []int{int(idx3)}, indices3)
+	require.DeepEqual(t, singleSig, att3.Signature)
+
+	// Add a different root
+	root2 := [32]byte{'s'}
+	msg.Data.BeaconBlockRoot = root2[:]
+	require.NoError(t, p.Add(msg, idx))
+	require.Equal(t, root2, p.root)
+	require.Equal(t, true, p.Seen(root2, idx))
+	require.Equal(t, false, p.Seen(root, idx))
+	att = p.attestations[primitives.PAYLOAD_PRESENT]
+	indices = att.AggregationBits.BitIndices()
+	require.DeepEqual(t, []int{int(idx)}, indices)
+}

beacon-chain/core/helpers/payload_attestation.go

@@ -68,6 +68,21 @@ func ValidateNilPayloadAttestation(att *eth.PayloadAttestation) error {
 	return ValidateNilPayloadAttestationData(att.Data)
 }
 
+// InPayloadTimelinessCommittee returns whether the given index belongs to the
+// PTC computed from the passed state.
+func InPayloadTimelinessCommittee(ctx context.Context, state state.ReadOnlyBeaconState, slot primitives.Slot, idx primitives.ValidatorIndex) (bool, error) {
+	ptc, err := GetPayloadTimelinessCommittee(ctx, state, slot)
+	if err != nil {
+		return false, err
+	}
+	for _, i := range ptc {
+		if i == idx {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // GetPayloadTimelinessCommittee returns the PTC for the given slot, computed from the passed state as in the
 // spec function `get_ptc`.
 func GetPayloadTimelinessCommittee(ctx context.Context, state state.ReadOnlyBeaconState, slot primitives.Slot) (indices []primitives.ValidatorIndex, err error) {
@@ -246,3 +261,36 @@ func IsValidIndexedPayloadAttestation(state state.ReadOnlyBeaconState, att *epbs
 
 	return signature.FastAggregateVerify(publicKeys, signingRoot), nil
 }
+
+// ValidatePayloadAttestationMessageSignature verifies the signature of a
+// payload attestation message.
+func ValidatePayloadAttestationMessageSignature(ctx context.Context, st state.ReadOnlyBeaconState, msg *eth.PayloadAttestationMessage) error {
+	if err := ValidateNilPayloadAttestationMessage(msg); err != nil {
+		return err
+	}
+	val, err := st.ValidatorAtIndex(msg.ValidatorIndex)
+	if err != nil {
+		return err
+	}
+	pub, err := bls.PublicKeyFromBytes(val.PublicKey)
+	if err != nil {
+		return err
+	}
+	sig, err := bls.SignatureFromBytes(msg.Signature)
+	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 := signing.ComputeSigningRoot(msg.Data, domain)
+	if err != nil {
+		return err
+	}
+	if !sig.Verify(pub, root[:]) {
+		return signing.ErrSigFailedToVerify
+	}
+	return nil
+}

beacon-chain/node/node.go

@@ -102,6 +102,7 @@ type BeaconNode struct {
 	blsToExecPool           blstoexec.PoolManager
 	depositCache            cache.DepositCache
 	trackedValidatorsCache  *cache.TrackedValidatorsCache
+	payloadAttestationCache *cache.PayloadAttestationCache
 	payloadIDCache          *cache.PayloadIDCache
 	stateFeed               *event.Feed
 	blockFeed               *event.Feed
@@ -152,6 +153,7 @@ func New(cliCtx *cli.Context, cancel context.CancelFunc, opts ...Option) (*Beaco
 		syncCommitteePool:       synccommittee.NewPool(),
 		blsToExecPool:           blstoexec.NewPool(),
 		trackedValidatorsCache:  cache.NewTrackedValidatorsCache(),
+		payloadAttestationCache: &cache.PayloadAttestationCache{},
 		payloadIDCache:          cache.NewPayloadIDCache(),
 		slasherBlockHeadersFeed: new(event.Feed),
 		slasherAttestationsFeed: new(event.Feed),
@@ -775,6 +777,7 @@ func (b *BeaconNode) registerBlockchainService(fc forkchoice.ForkChoicer, gs *st
 		blockchain.WithSyncComplete(syncComplete),
 		blockchain.WithBlobStorage(b.BlobStorage),
 		blockchain.WithTrackedValidatorsCache(b.trackedValidatorsCache),
+		blockchain.WithPayloadAttestationCache(b.payloadAttestationCache),
 		blockchain.WithPayloadIDCache(b.payloadIDCache),
 		blockchain.WithSyncChecker(b.syncChecker),
 	)
@@ -852,6 +855,7 @@ func (b *BeaconNode) registerSyncService(initialSyncComplete chan struct{}, bFil
 		regularsync.WithStateGen(b.stateGen),
 		regularsync.WithSlasherAttestationsFeed(b.slasherAttestationsFeed),
 		regularsync.WithSlasherBlockHeadersFeed(b.slasherBlockHeadersFeed),
+		regularsync.WithPayloadAttestationCache(b.payloadAttestationCache),
 		regularsync.WithPayloadReconstructor(web3Service),
 		regularsync.WithClockWaiter(b.clockWaiter),
 		regularsync.WithInitialSyncComplete(initialSyncComplete),

beacon-chain/sync/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "log.go",
         "metrics.go",
         "options.go",
+        "payload_attestations.go",
         "pending_attestations_queue.go",
         "pending_blocks_queue.go",
         "rate_limiter.go",
@@ -101,6 +102,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",
@@ -155,6 +157,7 @@ go_test(
         "decode_pubsub_test.go",
         "error_test.go",
         "fork_watcher_test.go",
+        "payload_attestations_test.go",
         "pending_attestations_queue_test.go",
         "pending_blocks_queue_test.go",
         "rate_limiter_test.go",
@@ -228,6 +231,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",
@@ -246,6 +250,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_d4l3k_messagediff//:go_default_library",