feat: a Gossipsub control plane with topic abstractions, a peer crawler and a peer controller

beacon-chain/p2p/BUILD.bazel

@@ -15,6 +15,8 @@ go_library(
         "fork_watcher.go",
         "gossip_scoring_params.go",
         "gossip_topic_mappings.go",
+        "gossipsub_peer_controller.go",
+        "gossipsub_peer_crawler.go",
         "handshake.go",
         "info.go",
         "interfaces.go",
@@ -51,6 +53,7 @@ go_library(
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/kv:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
+        "//beacon-chain/p2p/gossipsubcrawler:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/peerdata:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
@@ -115,6 +118,7 @@ go_library(
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_google_protobuf//proto:go_default_library",
+        "@org_golang_x_sync//semaphore:go_default_library",
     ],
 )
 
@@ -130,6 +134,8 @@ go_test(
         "fork_test.go",
         "gossip_scoring_params_test.go",
         "gossip_topic_mappings_test.go",
+        "gossipsub_peer_controller_test.go",
+        "gossipsub_peer_crawler_test.go",
         "message_id_test.go",
         "options_test.go",
         "parameter_test.go",
@@ -154,9 +160,11 @@ go_test(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
+        "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/iface:go_default_library",
         "//beacon-chain/db/testing:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
+        "//beacon-chain/p2p/gossipsubcrawler:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/peerdata:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
@@ -205,6 +213,7 @@ go_test(
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
         "@org_golang_google_protobuf//proto:go_default_library",
     ],
 )

beacon-chain/p2p/broadcaster.go

@@ -60,7 +60,10 @@ func (s *Service) Broadcast(ctx context.Context, msg proto.Message) error {
 	if !ok {
 		return errors.Errorf("message of %T does not support marshaller interface", msg)
 	}
-	return s.broadcastObject(ctx, castMsg, fmt.Sprintf(topic, forkDigest))
+
+	fullTopic := fmt.Sprintf(topic, forkDigest) + s.Encoding().ProtocolSuffix()
+
+	return s.broadcastObject(ctx, castMsg, fullTopic)
 }
 
 // BroadcastAttestation broadcasts an attestation to the p2p network, the message is assumed to be
@@ -106,6 +109,7 @@ func (s *Service) BroadcastSyncCommitteeMessage(ctx context.Context, subnet uint
 }
 
 func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint64, att ethpb.Att, forkDigest [fieldparams.VersionLength]byte) {
+	topic := AttestationSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.internalBroadcastAttestation")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -116,7 +120,7 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 
 	// Ensure we have peers with this subnet.
 	s.subnetLocker(subnet).RLock()
-	hasPeer := s.hasPeerWithSubnet(attestationToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(subnet).RUnlock()
 
 	span.SetAttributes(
@@ -131,7 +135,7 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 			s.subnetLocker(subnet).Lock()
 			defer s.subnetLocker(subnet).Unlock()
 
-			if err := s.FindAndDialPeersWithSubnets(ctx, AttestationSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipsubDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -153,13 +157,14 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 		return
 	}
 
-	if err := s.broadcastObject(ctx, att, attestationToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, att, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast attestation")
 		tracing.AnnotateError(span, err)
 	}
 }
 
 func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMsg *ethpb.SyncCommitteeMessage, forkDigest [fieldparams.VersionLength]byte) {
+	topic := SyncCommitteeSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.broadcastSyncCommittee")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -173,7 +178,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 	// to ensure that we can reuse the same subnet locker.
 	wrappedSubIdx := subnet + syncLockerVal
 	s.subnetLocker(wrappedSubIdx).RLock()
-	hasPeer := s.hasPeerWithSubnet(syncCommitteeToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(wrappedSubIdx).RUnlock()
 
 	span.SetAttributes(
@@ -187,7 +192,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 		if err := func() error {
 			s.subnetLocker(wrappedSubIdx).Lock()
 			defer s.subnetLocker(wrappedSubIdx).Unlock()
-			if err := s.FindAndDialPeersWithSubnets(ctx, SyncCommitteeSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipsubDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -205,7 +210,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 		return
 	}
 
-	if err := s.broadcastObject(ctx, sMsg, syncCommitteeToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, sMsg, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast sync committee message")
 		tracing.AnnotateError(span, err)
 	}
@@ -233,6 +238,7 @@ func (s *Service) BroadcastBlob(ctx context.Context, subnet uint64, blob *ethpb.
 }
 
 func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blobSidecar *ethpb.BlobSidecar, forkDigest [fieldparams.VersionLength]byte) {
+	topic := BlobSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.internalBroadcastBlob")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -243,7 +249,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 
 	wrappedSubIdx := subnet + blobSubnetLockerVal
 	s.subnetLocker(wrappedSubIdx).RLock()
-	hasPeer := s.hasPeerWithSubnet(blobSubnetToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(wrappedSubIdx).RUnlock()
 
 	if !hasPeer {
@@ -252,7 +258,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 			s.subnetLocker(wrappedSubIdx).Lock()
 			defer s.subnetLocker(wrappedSubIdx).Unlock()
 
-			if err := s.FindAndDialPeersWithSubnets(ctx, BlobSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipsubDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -264,7 +270,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 		}
 	}
 
-	if err := s.broadcastObject(ctx, blobSidecar, blobSubnetToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, blobSidecar, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast blob sidecar")
 		tracing.AnnotateError(span, err)
 	}
@@ -293,7 +299,7 @@ func (s *Service) BroadcastLightClientOptimisticUpdate(ctx context.Context, upda
 	}
 
 	digest := params.ForkDigest(slots.ToEpoch(update.AttestedHeader().Beacon().Slot))
-	if err := s.broadcastObject(ctx, update, lcOptimisticToTopic(digest)); err != nil {
+	if err := s.broadcastObject(ctx, update, LcOptimisticToTopic(digest)); err != nil {
 		log.WithError(err).Debug("Failed to broadcast light client optimistic update")
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
@@ -327,7 +333,7 @@ func (s *Service) BroadcastLightClientFinalityUpdate(ctx context.Context, update
 	}
 
 	forkDigest := params.ForkDigest(slots.ToEpoch(update.AttestedHeader().Beacon().Slot))
-	if err := s.broadcastObject(ctx, update, lcFinalityToTopic(forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, update, LcFinalityToTopic(forkDigest)); err != nil {
 		log.WithError(err).Debug("Failed to broadcast light client finality update")
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
@@ -386,13 +392,14 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 			subnet := peerdas.ComputeSubnetForDataColumnSidecar(sidecar.Index)
 
 			// Build the topic corresponding to subnet column subnet and this fork digest.
-			topic := dataColumnSubnetToTopic(subnet, forkDigest)
+			topic := DataColumnSubnetTopic(forkDigest, subnet)
 
 			// Compute the wrapped subnet index.
 			wrappedSubIdx := subnet + dataColumnSubnetVal
 
 			// Find peers if needed.
-			if err := s.findPeersIfNeeded(ctx, wrappedSubIdx, DataColumnSubnetTopicFormat, forkDigest, subnet); err != nil {
+
+			if err := s.findPeersIfNeeded(ctx, wrappedSubIdx, topic); err != nil {
 				tracing.AnnotateError(span, err)
 				log.WithError(err).Error("Cannot find peers if needed")
 				return
@@ -487,20 +494,16 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 func (s *Service) findPeersIfNeeded(
 	ctx context.Context,
 	wrappedSubIdx uint64,
-	topicFormat string,
-	forkDigest [fieldparams.VersionLength]byte,
-	subnet uint64,
+	topic string,
 ) error {
 	// Sending a data column sidecar to only one peer is not ideal,
 	// but it ensures at least one peer receives it.
 	s.subnetLocker(wrappedSubIdx).Lock()
 	defer s.subnetLocker(wrappedSubIdx).Unlock()
 
-	// No peers found, attempt to find peers with this subnet.
-	if err := s.FindAndDialPeersWithSubnets(ctx, topicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+	if err := s.gossipsubDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 		return errors.Wrap(err, "find peers with subnet")
 	}
-
 	return nil
 }
 
@@ -525,34 +528,10 @@ func (s *Service) broadcastObject(ctx context.Context, obj ssz.Marshaler, topic
 		iid := int64(id)
 		span = trace.AddMessageSendEvent(span, iid, messageLen /*uncompressed*/, messageLen /*compressed*/)
 	}
-	if err := s.PublishToTopic(ctx, topic+s.Encoding().ProtocolSuffix(), buf.Bytes()); err != nil {
+	if err := s.PublishToTopic(ctx, topic, buf.Bytes()); err != nil {
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
 		return err
 	}
 	return nil
 }
-
-func attestationToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(AttestationSubnetTopicFormat, forkDigest, subnet)
-}
-
-func syncCommitteeToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(SyncCommitteeSubnetTopicFormat, forkDigest, subnet)
-}
-
-func blobSubnetToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(BlobSubnetTopicFormat, forkDigest, subnet)
-}
-
-func lcOptimisticToTopic(forkDigest [4]byte) string {
-	return fmt.Sprintf(LightClientOptimisticUpdateTopicFormat, forkDigest)
-}
-
-func lcFinalityToTopic(forkDigest [4]byte) string {
-	return fmt.Sprintf(LightClientFinalityUpdateTopicFormat, forkDigest)
-}
-
-func dataColumnSubnetToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(DataColumnSubnetTopicFormat, forkDigest, subnet)
-}

beacon-chain/p2p/broadcaster_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 	"github.com/OffchainLabs/prysm/v7/testing/util"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
+	"github.com/ethereum/go-ethereum/p2p/enode"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/host"
 	"google.golang.org/protobuf/proto"
@@ -226,6 +227,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 	// Setup bootnode.
 	cfg := &Config{PingInterval: testPingInterval, DB: db}
 	cfg.UDPPort = uint(port)
+	cfg.TCPPort = uint(port)
 	_, pkey := createAddrAndPrivKey(t)
 	ipAddr := net.ParseIP("127.0.0.1")
 	genesisTime := time.Now()
@@ -251,8 +253,9 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 
 	var listeners []*listenerWrapper
 	var hosts []host.Host
+	var configs []*Config
 	// setup other nodes.
-	cfg = &Config{
+	baseCfg := &Config{
 		Discv5BootStrapAddrs: []string{bootNode.String()},
 		MaxPeers:             2,
 		PingInterval:         testPingInterval,
@@ -261,11 +264,21 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 	// Setup 2 different hosts
 	for i := uint(1); i <= 2; i++ {
 		h, pkey, ipAddr := createHost(t, port+i)
-		cfg.UDPPort = uint(port + i)
-		cfg.TCPPort = uint(port + i)
+
+		// Create a new config for each service to avoid shared mutations
+		cfg := &Config{
+			Discv5BootStrapAddrs: baseCfg.Discv5BootStrapAddrs,
+			MaxPeers:             baseCfg.MaxPeers,
+			PingInterval:         baseCfg.PingInterval,
+			DB:                   baseCfg.DB,
+			UDPPort:              uint(port + i),
+			TCPPort:              uint(port + i),
+		}
+
 		if len(listeners) > 0 {
 			cfg.Discv5BootStrapAddrs = append(cfg.Discv5BootStrapAddrs, listeners[len(listeners)-1].Self().String())
 		}
+
 		s := &Service{
 			cfg:                   cfg,
 			genesisTime:           genesisTime,
@@ -278,18 +291,22 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		close(s.custodyInfoSet)
 
 		listener, err := s.startDiscoveryV5(ipAddr, pkey)
-		// Set for 2nd peer
+		assert.NoError(t, err, "Could not start discovery for node")
+
+		// Set listener for the service
+		s.dv5Listener = listener
+		s.metaData = wrapper.WrappedMetadataV0(new(ethpb.MetaDataV0))
+
+		// Set subnet for 2nd peer
 		if i == 2 {
-			s.dv5Listener = listener
-			s.metaData = wrapper.WrappedMetadataV0(new(ethpb.MetaDataV0))
 			bitV := bitfield.NewBitvector64()
 			bitV.SetBitAt(subnet, true)
 			err := s.updateSubnetRecordWithMetadata(bitV)
 			require.NoError(t, err)
 		}
-		assert.NoError(t, err, "Could not start discovery for node")
 		listeners = append(listeners, listener)
 		hosts = append(hosts, h)
+		configs = append(configs, cfg)
 	}
 	defer func() {
 		// Close down all peers.
@@ -324,7 +341,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		pubsub:                ps1,
 		dv5Listener:           listeners[0],
 		joinedTopics:          map[string]*pubsub.Topic{},
-		cfg:                   cfg,
+		cfg:                   configs[0],
 		genesisTime:           time.Now(),
 		genesisValidatorsRoot: bytesutil.PadTo([]byte{'A'}, 32),
 		subnetsLock:           make(map[uint64]*sync.RWMutex),
@@ -340,7 +357,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		pubsub:                ps2,
 		dv5Listener:           listeners[1],
 		joinedTopics:          map[string]*pubsub.Topic{},
-		cfg:                   cfg,
+		cfg:                   configs[1],
 		genesisTime:           time.Now(),
 		genesisValidatorsRoot: bytesutil.PadTo([]byte{'A'}, 32),
 		subnetsLock:           make(map[uint64]*sync.RWMutex),
@@ -762,6 +779,16 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	subnet := peerdas.ComputeSubnetForDataColumnSidecar(columnIndex)
 	topic := fmt.Sprintf(topicFormat, digest, subnet) + service.Encoding().ProtocolSuffix()
 
+	crawler, err := NewGossipsubPeerCrawler(service, listener, 1*time.Second, 1*time.Second, 10,
+		func(n *enode.Node) bool { return true },
+		service.Peers().Scorers().Score)
+	require.NoError(t, err)
+	err = crawler.Start(func(ctx context.Context, node *enode.Node) ([]string, error) {
+		return []string{topic}, nil
+	})
+	require.NoError(t, err)
+	service.gossipsubDialer = NewGossipsubPeerDialer(crawler, service.PubSub().ListPeers, service.DialPeers)
+
 	_, verifiedRoSidecars := util.CreateTestVerifiedRoDataColumnSidecars(t, []util.DataColumnParam{{Index: columnIndex}})
 	verifiedRoSidecar := verifiedRoSidecars[0]
 
@@ -787,3 +814,7 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	require.NoError(t, service.Encoding().DecodeGossip(msg.Data, &result))
 	require.DeepEqual(t, &result, verifiedRoSidecar)
 }
+
+func attestationToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
+	return fmt.Sprintf(AttestationSubnetTopicFormat, forkDigest, subnet)
+}