Add Gossip Handler For BLS To Execution Changes

beacon-chain/node/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
         "//beacon-chain/monitor:go_default_library",
         "//beacon-chain/node/registration:go_default_library",
         "//beacon-chain/operations/attestations:go_default_library",
+        "//beacon-chain/operations/blstoexec:go_default_library",
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/operations/synccommittee:go_default_library",
         "//beacon-chain/operations/voluntaryexits:go_default_library",

beacon-chain/node/node.go

@@ -34,6 +34,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/monitor"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/node/registration"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/attestations"
+	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/blstoexec"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/slashings"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/synccommittee"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/voluntaryexits"
@@ -94,6 +95,7 @@ type BeaconNode struct {
 	exitPool                voluntaryexits.PoolManager
 	slashingsPool           slashings.PoolManager
 	syncCommitteePool       synccommittee.Pool
+	blsToExecPool           blstoexec.PoolManager
 	depositCache            *depositcache.DepositCache
 	proposerIdsCache        *cache.ProposerPayloadIDsCache
 	stateFeed               *event.Feed
@@ -171,6 +173,7 @@ func New(cliCtx *cli.Context, opts ...Option) (*BeaconNode, error) {
 		exitPool:                voluntaryexits.NewPool(),
 		slashingsPool:           slashings.NewPool(),
 		syncCommitteePool:       synccommittee.NewPool(),
+		blsToExecPool:           blstoexec.NewPool(),
 		slasherBlockHeadersFeed: new(event.Feed),
 		slasherAttestationsFeed: new(event.Feed),
 		serviceFlagOpts:         &serviceFlagOpts{},
@@ -674,6 +677,7 @@ func (b *BeaconNode) registerSyncService() error {
 		regularsync.WithExitPool(b.exitPool),
 		regularsync.WithSlashingPool(b.slashingsPool),
 		regularsync.WithSyncCommsPool(b.syncCommitteePool),
+		regularsync.WithBlsToExecPool(b.blsToExecPool),
 		regularsync.WithStateGen(b.stateGen),
 		regularsync.WithSlasherAttestationsFeed(b.slasherAttestationsFeed),
 		regularsync.WithSlasherBlockHeadersFeed(b.slasherBlockHeadersFeed),

beacon-chain/operations/blstoexec/pool.go

@@ -17,6 +17,7 @@ type PoolManager interface {
 	BLSToExecChangesForInclusion() ([]*ethpb.SignedBLSToExecutionChange, error)
 	InsertBLSToExecChange(change *ethpb.SignedBLSToExecutionChange)
 	MarkIncluded(change *ethpb.SignedBLSToExecutionChange) error
+	ValidatorExists(idx types.ValidatorIndex) bool
 }
 
 // Pool is a concrete implementation of PoolManager.
@@ -107,3 +108,14 @@ func (p *Pool) MarkIncluded(change *ethpb.SignedBLSToExecutionChange) error {
 	p.pending.Remove(node)
 	return nil
 }
+
+// ValidatorExists checks if the bls to execution change object exists
+// for that particular validator.
+func (p *Pool) ValidatorExists(idx types.ValidatorIndex) bool {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	node := p.m[idx]
+
+	return node != nil
+}

beacon-chain/p2p/gossip_scoring_params.go

@@ -116,6 +116,8 @@ func (s *Service) topicScoreParams(topic string) (*pubsub.TopicScoreParams, erro
 		return defaultProposerSlashingTopicParams(), nil
 	case strings.Contains(topic, GossipAttesterSlashingMessage):
 		return defaultAttesterSlashingTopicParams(), nil
+	case strings.Contains(topic, GossipBlsToExecutionChangeMessage):
+		return defaultBlsToExecutionChangeTopicParams(), nil
 	default:
 		return nil, errors.Errorf("unrecognized topic provided for parameter registration: %s", topic)
 	}
@@ -473,6 +475,28 @@ func defaultVoluntaryExitTopicParams() *pubsub.TopicScoreParams {
 	}
 }
 
+func defaultBlsToExecutionChangeTopicParams() *pubsub.TopicScoreParams {
+	return &pubsub.TopicScoreParams{
+		TopicWeight:                     voluntaryExitWeight,
+		TimeInMeshWeight:                maxInMeshScore / inMeshCap(),
+		TimeInMeshQuantum:               inMeshTime(),
+		TimeInMeshCap:                   inMeshCap(),
+		FirstMessageDeliveriesWeight:    2,
+		FirstMessageDeliveriesDecay:     scoreDecay(oneHundredEpochs),
+		FirstMessageDeliveriesCap:       5,
+		MeshMessageDeliveriesWeight:     0,
+		MeshMessageDeliveriesDecay:      0,
+		MeshMessageDeliveriesCap:        0,
+		MeshMessageDeliveriesThreshold:  0,
+		MeshMessageDeliveriesWindow:     0,
+		MeshMessageDeliveriesActivation: 0,
+		MeshFailurePenaltyWeight:        0,
+		MeshFailurePenaltyDecay:         0,
+		InvalidMessageDeliveriesWeight:  -2000,
+		InvalidMessageDeliveriesDecay:   scoreDecay(invalidDecayPeriod),
+	}
+}
+
 func oneSlotDuration() time.Duration {
 	return time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second
 }

beacon-chain/p2p/gossip_topic_mappings.go

@@ -20,6 +20,7 @@ var gossipTopicMappings = map[string]proto.Message{
 	AggregateAndProofSubnetTopicFormat:        &ethpb.SignedAggregateAttestationAndProof{},
 	SyncContributionAndProofSubnetTopicFormat: &ethpb.SignedContributionAndProof{},
 	SyncCommitteeSubnetTopicFormat:            &ethpb.SyncCommitteeMessage{},
+	BlsToExecutionChangeSubnetTopicFormat:     &ethpb.SignedBLSToExecutionChange{},
 }
 
 // GossipTopicMappings is a function to return the assigned data type

beacon-chain/p2p/topics.go

@@ -26,6 +26,8 @@ const (
 	GossipAggregateAndProofMessage = "beacon_aggregate_and_proof"
 	// GossipContributionAndProofMessage is the name for the sync contribution and proof message type.
 	GossipContributionAndProofMessage = "sync_committee_contribution_and_proof"
+	// GossipBlsToExecutionChangeMessage is the name for the bls to execution change message type.
+	GossipBlsToExecutionChangeMessage = "bls_to_execution_change"
 
 	// Topic Formats
 	//
@@ -45,4 +47,6 @@ const (
 	AggregateAndProofSubnetTopicFormat = GossipProtocolAndDigest + GossipAggregateAndProofMessage
 	// SyncContributionAndProofSubnetTopicFormat is the topic format for the sync aggregate and proof subnet.
 	SyncContributionAndProofSubnetTopicFormat = GossipProtocolAndDigest + GossipContributionAndProofMessage
+	// BlsToExecutionChangeSubnetTopicFormat is the topic format for the bls to execution change subnet.
+	BlsToExecutionChangeSubnetTopicFormat = GossipProtocolAndDigest + GossipBlsToExecutionChangeMessage
 )

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

@@ -6,6 +6,7 @@ go_library(
         "assignments.go",
         "attestations.go",
         "blocks.go",
+        "blstoexec.go",
         "committees.go",
         "config.go",
         "log.go",

beacon-chain/sync/BUILD.bazel

@@ -31,6 +31,7 @@ go_library(
         "subscriber_beacon_aggregate_proof.go",
         "subscriber_beacon_attestation.go",
         "subscriber_beacon_blocks.go",
+        "subscriber_bls_to_execution_change.go",
         "subscriber_handlers.go",
         "subscriber_sync_committee_message.go",
         "subscriber_sync_contribution_proof.go",
@@ -40,6 +41,7 @@ go_library(
         "validate_attester_slashing.go",
         "validate_beacon_attestation.go",
         "validate_beacon_blocks.go",
+        "validate_bls_to_execution_change.go",
         "validate_proposer_slashing.go",
         "validate_sync_committee_message.go",
         "validate_sync_contribution_proof.go",
@@ -71,6 +73,7 @@ go_library(
         "//beacon-chain/db/filters:go_default_library",
         "//beacon-chain/execution:go_default_library",
         "//beacon-chain/operations/attestations:go_default_library",
+        "//beacon-chain/operations/blstoexec:go_default_library",
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/operations/synccommittee:go_default_library",
         "//beacon-chain/operations/voluntaryexits:go_default_library",
@@ -158,6 +161,7 @@ go_test(
         "validate_attester_slashing_test.go",
         "validate_beacon_attestation_test.go",
         "validate_beacon_blocks_test.go",
+        "validate_bls_to_execution_change_test.go",
         "validate_proposer_slashing_test.go",
         "validate_sync_committee_message_test.go",
         "validate_sync_contribution_proof_test.go",
@@ -185,6 +189,7 @@ go_test(
         "//beacon-chain/execution/testing:go_default_library",
         "//beacon-chain/forkchoice/doubly-linked-tree:go_default_library",
         "//beacon-chain/operations/attestations:go_default_library",
+        "//beacon-chain/operations/blstoexec:go_default_library",
         "//beacon-chain/operations/slashings:go_default_library",
         "//beacon-chain/p2p:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",

beacon-chain/sync/fork_watcher.go

@@ -49,26 +49,16 @@ func (s *Service) registerForUpcomingFork(currEpoch types.Epoch) error {
 	// will subscribe the new topics in advance.
 	if isNextForkEpoch {
 		nextEpoch := currEpoch + 1
-		switch nextEpoch {
-		case params.BeaconConfig().AltairForkEpoch:
-			digest, err := forks.ForkDigestFromEpoch(nextEpoch, genRoot[:])
-			if err != nil {
-				return errors.Wrap(err, "Could not retrieve fork digest")
-			}
-			if s.subHandler.digestExists(digest) {
-				return nil
-			}
-			s.registerSubscribers(nextEpoch, digest)
+		digest, err := forks.ForkDigestFromEpoch(nextEpoch, genRoot[:])
+		if err != nil {
+			return errors.Wrap(err, "could not retrieve fork digest")
+		}
+		if s.subHandler.digestExists(digest) {
+			return nil
+		}
+		s.registerSubscribers(nextEpoch, digest)
+		if nextEpoch == params.BeaconConfig().AltairForkEpoch {
 			s.registerRPCHandlersAltair()
-		case params.BeaconConfig().BellatrixForkEpoch:
-			digest, err := forks.ForkDigestFromEpoch(nextEpoch, genRoot[:])
-			if err != nil {
-				return errors.Wrap(err, "could not retrieve fork digest")
-			}
-			if s.subHandler.digestExists(digest) {
-				return nil
-			}
-			s.registerSubscribers(nextEpoch, digest)
 		}
 	}
 	return nil
@@ -109,9 +99,7 @@ func (s *Service) deregisterFromPastFork(currEpoch types.Epoch) error {
 		if err != nil {
 			return errors.Wrap(err, "failed to determine previous epoch fork data")
 		}
-
-		switch prevFork.Epoch {
-		case params.BeaconConfig().GenesisEpoch:
+		if prevFork.Epoch == params.BeaconConfig().GenesisEpoch {
 			s.unregisterPhase0Handlers()
 		}
 		// Run through all our current active topics and see

beacon-chain/sync/options.go

@@ -8,6 +8,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/execution"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/attestations"
+	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/blstoexec"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/slashings"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/synccommittee"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/voluntaryexits"
@@ -66,6 +67,13 @@ func WithSyncCommsPool(syncCommsPool synccommittee.Pool) Option {
 	}
 }
 
+func WithBlsToExecPool(blsToExecPool blstoexec.PoolManager) Option {
+	return func(s *Service) error {
+		s.cfg.blsToExecPool = blsToExecPool
+		return nil
+	}
+}
+
 func WithChainService(chain blockchainService) Option {
 	return func(s *Service) error {
 		s.cfg.chain = chain

beacon-chain/sync/service.go

@@ -26,6 +26,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/execution"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/attestations"
+	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/blstoexec"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/slashings"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/synccommittee"
 	"github.com/prysmaticlabs/prysm/v3/beacon-chain/operations/voluntaryexits"
@@ -75,6 +76,7 @@ type config struct {
 	exitPool                      voluntaryexits.PoolManager
 	slashingPool                  slashings.PoolManager
 	syncCommsPool                 synccommittee.Pool
+	blsToExecPool                 blstoexec.PoolManager
 	chain                         blockchainService
 	initialSync                   Checker
 	stateNotifier                 statefeed.Notifier

beacon-chain/sync/subscriber.go

@@ -120,6 +120,16 @@ func (s *Service) registerSubscribers(epoch types.Epoch, digest [4]byte) {
 			)
 		}
 	}
+
+	// New Gossip Topic in Capella
+	if epoch >= params.BeaconConfig().CapellaForkEpoch {
+		s.subscribe(
+			p2p.BlsToExecutionChangeSubnetTopicFormat,
+			s.validateBlsToExecutionChange,
+			s.blsToExecutionChangeSubscriber,
+			digest,
+		)
+	}
 }
 
 // subscribe to a given topic with a given validator and subscription handler.

beacon-chain/sync/subscriber_bls_to_execution_change.go

@@ -0,0 +1,18 @@
+package sync
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	ethpb "github.com/prysmaticlabs/prysm/v3/proto/prysm/v1alpha1"
+	"google.golang.org/protobuf/proto"
+)
+
+func (s *Service) blsToExecutionChangeSubscriber(ctx context.Context, msg proto.Message) error {
+	blsMsg, ok := msg.(*ethpb.SignedBLSToExecutionChange)
+	if !ok {
+		return errors.Errorf("incorrect type of message received, wanted %T but got %T", &ethpb.SignedBLSToExecutionChange{}, msg)
+	}
+	s.cfg.blsToExecPool.InsertBLSToExecChange(blsMsg)
+	return nil
+}

beacon-chain/sync/validate_bls_to_execution_change.go

@@ -0,0 +1,63 @@
+package sync
+
+import (
+	"context"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/prysmaticlabs/prysm/v3/beacon-chain/core/signing"
+	"github.com/prysmaticlabs/prysm/v3/config/params"
+	"github.com/prysmaticlabs/prysm/v3/monitoring/tracing"
+	ethpb "github.com/prysmaticlabs/prysm/v3/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v3/time/slots"
+	"go.opencensus.io/trace"
+)
+
+func (s *Service) validateBlsToExecutionChange(ctx context.Context, pid peer.ID, msg *pubsub.Message) (pubsub.ValidationResult, error) {
+	// Validation runs on publish (not just subscriptions), so we should approve any message from
+	// ourselves.
+	if pid == s.cfg.p2p.PeerID() {
+		return pubsub.ValidationAccept, nil
+	}
+
+	// The head state will be too far away to validate any execution change.
+	if s.cfg.initialSync.Syncing() {
+		return pubsub.ValidationIgnore, nil
+	}
+
+	ctx, span := trace.StartSpan(ctx, "sync.validateBlsToExecutionChange")
+	defer span.End()
+
+	m, err := s.decodePubsubMessage(msg)
+	if err != nil {
+		tracing.AnnotateError(span, err)
+		return pubsub.ValidationReject, err
+	}
+
+	blsChange, ok := m.(*ethpb.SignedBLSToExecutionChange)
+	if !ok {
+		return pubsub.ValidationReject, errWrongMessage
+	}
+
+	if s.cfg.blsToExecPool.ValidatorExists(blsChange.Message.ValidatorIndex) {
+		return pubsub.ValidationIgnore, nil
+	}
+	st, err := s.cfg.chain.HeadState(ctx)
+	if err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+
+	epoch := slots.ToEpoch(st.Slot())
+	domain, err := signing.Domain(st.Fork(), epoch, params.BeaconConfig().DomainBLSToExecutionChange, st.GenesisValidatorsRoot())
+	if err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+	if err := signing.VerifySigningRoot(blsChange.Message, blsChange.Message.FromBlsPubkey, blsChange.Signature, domain); err != nil {
+		return pubsub.ValidationReject, signing.ErrSigFailedToVerify
+	}
+	// TODO(Potuz): BLSChange Validation
+	// TODO(Nishant): Add to batch gossip sig verification
+
+	msg.ValidatorData = blsChange // Used in downstream subscriber
+	return pubsub.ValidationAccept, nil
+}