Fix Initial Sync with 128 data columns subnets

beacon-chain/p2p/discovery.go

@@ -119,7 +119,7 @@ func (s *Service) RefreshPersistentSubnets() {
 		s.updateSubnetRecordWithMetadata(bitV)
 
 		// Ping all peers.
-		s.pingPeers()
+		s.pingPeersAndLogEnr()
 
 		return
 	}
@@ -157,7 +157,7 @@ func (s *Service) RefreshPersistentSubnets() {
 		s.updateSubnetRecordWithMetadataV2(bitV, bitS)
 
 		// Ping all peers to inform them of new metadata
-		s.pingPeers()
+		s.pingPeersAndLogEnr()
 
 		return
 	}
@@ -186,7 +186,7 @@ func (s *Service) RefreshPersistentSubnets() {
 	s.updateSubnetRecordWithMetadataV3(bitV, bitS, custodySubnetCount)
 
 	// Ping all peers.
-	s.pingPeers()
+	s.pingPeersAndLogEnr()
 }
 
 // listen for new nodes watches for new nodes in the network and adds them to the peerstore.

beacon-chain/p2p/handshake.go

@@ -90,14 +90,15 @@ func (s *Service) AddConnectionHandler(reqFunc, goodByeFunc func(ctx context.Con
 					disconnectFromPeer()
 					return
 				}
+
 				validPeerConnection := func() {
 					s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerConnected)
 					// Go through the handshake process.
 					log.WithFields(logrus.Fields{
-						"direction":   conn.Stat().Direction,
+						"direction":   conn.Stat().Direction.String(),
 						"multiAddr":   peerMultiaddrString(conn),
 						"activePeers": len(s.peers.Active()),
-					}).Debug("Peer connected")
+					}).Debug("New peer connection")
 				}
 
 				// Do not perform handshake on inbound dials.
@@ -173,7 +174,11 @@ func (s *Service) AddDisconnectionHandler(handler func(ctx context.Context, id p
 				s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerDisconnected)
 				// Only log disconnections if we were fully connected.
 				if priorState == peers.PeerConnected {
-					log.WithField("activePeers", len(s.peers.Active())).Debug("Peer disconnected")
+					log.WithFields(logrus.Fields{
+						"direction":   conn.Stat().Direction.String(),
+						"multiAddr":   peerMultiaddrString(conn),
+						"activePeers": len(s.peers.Active()),
+					}).Debug("Peer disconnected")
 				}
 			}()
 		},

beacon-chain/p2p/pubsub_filter.go

@@ -10,16 +10,27 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/encoder"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/network/forks"
+	"github.com/sirupsen/logrus"
 )
 
 var _ pubsub.SubscriptionFilter = (*Service)(nil)
 
 // It is set at this limit to handle the possibility
 // of double topic subscriptions at fork boundaries.
-// -> 64 Attestation Subnets * 2.
-// -> 4 Sync Committee Subnets * 2.
-// -> Block,Aggregate,ProposerSlashing,AttesterSlashing,Exits,SyncContribution * 2.
-const pubsubSubscriptionRequestLimit = 200
+// -> BeaconBlock              * 2 = 2
+// -> BeaconAggregateAndProof  * 2 = 2
+// -> VoluntaryExit            * 2 = 2
+// -> ProposerSlashing         * 2 = 2
+// -> AttesterSlashing         * 2 = 2
+// -> 64 Beacon Attestation    * 2 = 128
+// -> SyncContributionAndProof * 2 = 2
+// -> 4 SyncCommitteeSubnets   * 2 = 8
+// -> BlsToExecutionChange     * 2 = 2
+// -> 128 DataColumnSidecar    * 2 = 256
+// -------------------------------------
+// TOTAL                           = 406
+// (Note: BlobSidecar is not included in this list since it is superseded by DataColumnSidecar)
+const pubsubSubscriptionRequestLimit = 500
 
 // CanSubscribe returns true if the topic is of interest and we could subscribe to it.
 func (s *Service) CanSubscribe(topic string) bool {
@@ -95,8 +106,15 @@ func (s *Service) CanSubscribe(topic string) bool {
 // FilterIncomingSubscriptions is invoked for all RPCs containing subscription notifications.
 // This method returns only the topics of interest and may return an error if the subscription
 // request contains too many topics.
-func (s *Service) FilterIncomingSubscriptions(_ peer.ID, subs []*pubsubpb.RPC_SubOpts) ([]*pubsubpb.RPC_SubOpts, error) {
+func (s *Service) FilterIncomingSubscriptions(peerID peer.ID, subs []*pubsubpb.RPC_SubOpts) ([]*pubsubpb.RPC_SubOpts, error) {
 	if len(subs) > pubsubSubscriptionRequestLimit {
+		subsCount := len(subs)
+		log.WithFields(logrus.Fields{
+			"peerID":             peerID,
+			"subscriptionCounts": subsCount,
+			"subscriptionLimit":  pubsubSubscriptionRequestLimit,
+		}).Debug("Too many incoming subscriptions, filtering them")
+
 		return nil, pubsub.ErrTooManySubscriptions
 	}
 

beacon-chain/p2p/service.go

@@ -202,12 +202,13 @@ func (s *Service) Start() {
 			s.startupErr = err
 			return
 		}
-		err = s.connectToBootnodes()
-		if err != nil {
-			log.WithError(err).Error("Could not add bootnode to the exclusion list")
+
+		if err := s.connectToBootnodes(); err != nil {
+			log.WithError(err).Error("Could not connect to boot nodes")
 			s.startupErr = err
 			return
 		}
+
 		s.dv5Listener = listener
 		go s.listenForNewNodes()
 	}
@@ -393,12 +394,17 @@ func (s *Service) AddPingMethod(reqFunc func(ctx context.Context, id peer.ID) er
 	s.pingMethodLock.Unlock()
 }
 
-func (s *Service) pingPeers() {
+func (s *Service) pingPeersAndLogEnr() {
 	s.pingMethodLock.RLock()
 	defer s.pingMethodLock.RUnlock()
+
+	localENR := s.dv5Listener.Self()
+	log.WithField("ENR", localENR).Info("New node record")
+
 	if s.pingMethod == nil {
 		return
 	}
+
 	for _, pid := range s.peers.Connected() {
 		go func(id peer.ID) {
 			if err := s.pingMethod(s.ctx, id); err != nil {

beacon-chain/p2p/subnets.go

@@ -2,6 +2,7 @@ package p2p
 
 import (
 	"context"
+	"math"
 	"strings"
 	"sync"
 	"time"
@@ -20,9 +21,9 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/wrapper"
 	"github.com/prysmaticlabs/prysm/v5/crypto/hash"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
-	mathutil "github.com/prysmaticlabs/prysm/v5/math"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	pb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/sirupsen/logrus"
 )
 
 var attestationSubnetCount = params.BeaconConfig().AttestationSubnetCount
@@ -53,6 +54,79 @@ const blobSubnetLockerVal = 110
 // chosen more than sync, attestation and blob subnet (6) combined.
 const dataColumnSubnetVal = 150
 
+// nodeFilter return a function that filters nodes based on the subnet topic and subnet index.
+func (s *Service) nodeFilter(topic string, index uint64) (func(node *enode.Node) bool, error) {
+	switch {
+	case strings.Contains(topic, GossipAttestationMessage):
+		return s.filterPeerForAttSubnet(index), nil
+	case strings.Contains(topic, GossipSyncCommitteeMessage):
+		return s.filterPeerForSyncSubnet(index), nil
+	case strings.Contains(topic, GossipDataColumnSidecarMessage):
+		return s.filterPeerForDataColumnsSubnet(index), nil
+	default:
+		return nil, errors.Errorf("no subnet exists for provided topic: %s", topic)
+	}
+}
+
+// searchForPeers performs a network search for peers subscribed to a particular subnet.
+// It exits as soon as one of these conditions is met:
+// - It looped through `batchSize` nodes.
+// - It found `peersToFindCount“ peers corresponding to the `filter` criteria.
+// - Iterator is exhausted.
+func searchForPeers(
+	iterator enode.Iterator,
+	batchSize int,
+	peersToFindCount int,
+	filter func(node *enode.Node) bool,
+) []*enode.Node {
+	nodeFromNodeID := make(map[enode.ID]*enode.Node, batchSize)
+	for i := 0; i < batchSize && len(nodeFromNodeID) <= peersToFindCount && iterator.Next(); i++ {
+		node := iterator.Node()
+
+		// Filter out nodes that do not meet the criteria.
+		if !filter(node) {
+			continue
+		}
+
+		// Remove duplicates, keeping the node with higher seq.
+		prevNode, ok := nodeFromNodeID[node.ID()]
+		if ok && prevNode.Seq() > node.Seq() {
+			continue
+		}
+
+		nodeFromNodeID[node.ID()] = node
+	}
+
+	// Convert the map to a slice.
+	nodes := make([]*enode.Node, 0, len(nodeFromNodeID))
+	for _, node := range nodeFromNodeID {
+		nodes = append(nodes, node)
+	}
+
+	return nodes
+}
+
+// dialPeer dials a peer in a separate goroutine.
+func (s *Service) dialPeer(ctx context.Context, wg *sync.WaitGroup, node *enode.Node) {
+	info, _, err := convertToAddrInfo(node)
+	if err != nil {
+		return
+	}
+
+	if info == nil {
+		return
+	}
+
+	wg.Add(1)
+	go func() {
+		if err := s.connectWithPeer(ctx, *info); err != nil {
+			log.WithError(err).Tracef("Could not connect with peer %s", info.String())
+		}
+
+		wg.Done()
+	}()
+}
+
 // FindPeersWithSubnet performs a network search for peers
 // subscribed to a particular subnet. Then it tries to connect
 // with those peers. This method will block until either:
@@ -61,69 +135,96 @@ const dataColumnSubnetVal = 150
 // On some edge cases, this method may hang indefinitely while peers
 // are actually found. In such a case, the user should cancel the context
 // and re-run the method again.
-func (s *Service) FindPeersWithSubnet(ctx context.Context, topic string,
-	index uint64, threshold int) (bool, error) {
+func (s *Service) FindPeersWithSubnet(
+	ctx context.Context,
+	topic string,
+	index uint64,
+	threshold int,
+) (bool, error) {
+	const batchSize = 2000
+
 	ctx, span := trace.StartSpan(ctx, "p2p.FindPeersWithSubnet")
 	defer span.End()
 
 	span.AddAttributes(trace.Int64Attribute("index", int64(index))) // lint:ignore uintcast -- It's safe to do this for tracing.
 
 	if s.dv5Listener == nil {
-		// return if discovery isn't set
+		// Return if discovery isn't set
 		return false, nil
 	}
 
 	topic += s.Encoding().ProtocolSuffix()
 	iterator := s.dv5Listener.RandomNodes()
 	defer iterator.Close()
-	switch {
-	case strings.Contains(topic, GossipAttestationMessage):
-		iterator = filterNodes(ctx, iterator, s.filterPeerForAttSubnet(index))
-	case strings.Contains(topic, GossipSyncCommitteeMessage):
-		iterator = filterNodes(ctx, iterator, s.filterPeerForSyncSubnet(index))
-	case strings.Contains(topic, GossipDataColumnSidecarMessage):
-		iterator = filterNodes(ctx, iterator, s.filterPeerForDataColumnsSubnet(index))
-	default:
-		return false, errors.Errorf("no subnet exists for provided topic: %s", topic)
+
+	filter, err := s.nodeFilter(topic, index)
+	if err != nil {
+		return false, errors.Wrap(err, "node filter")
+	}
+
+	peersSummary := func(topic string, threshold int) (int, int) {
+		// Retrieve how many peers we have for this topic.
+		peerCountForTopic := len(s.pubsub.ListPeers(topic))
+
+		// Compute how many peers we are missing to reach the threshold.
+		missingPeerCountForTopic := max(0, threshold-peerCountForTopic)
+
+		return peerCountForTopic, missingPeerCountForTopic
 	}
 
+	// Compute how many peers we are missing to reach the threshold.
+	peerCountForTopic, missingPeerCountForTopic := peersSummary(topic, threshold)
+
+	// Exit early if we have enough peers.
+	if missingPeerCountForTopic == 0 {
+		return true, nil
+	}
+
+	log.WithFields(logrus.Fields{
+		"topic":            topic,
+		"currentPeerCount": peerCountForTopic,
+		"targetPeerCount":  threshold,
+	}).Debug("Searching for new peers in the network - Start")
+
 	wg := new(sync.WaitGroup)
 	for {
-		currNum := len(s.pubsub.ListPeers(topic))
-		if currNum >= threshold {
+		// If we have enough peers, we can exit the loop. This is the happy path.
+		if missingPeerCountForTopic == 0 {
 			break
 		}
+
+		// If the context is done, we can exit the loop. This is the unhappy path.
 		if err := ctx.Err(); err != nil {
-			return false, errors.Errorf("unable to find requisite number of peers for topic %s - "+
-				"only %d out of %d peers were able to be found", topic, currNum, threshold)
+			return false, errors.Errorf(
+				"unable to find requisite number of peers for topic %s - only %d out of %d peers available after searching",
+				topic, peerCountForTopic, threshold,
+			)
 		}
-		nodeCount := int(params.BeaconNetworkConfig().MinimumPeersInSubnetSearch)
+
+		// Search for new peers in the network.
+		nodes := searchForPeers(iterator, batchSize, missingPeerCountForTopic, filter)
+
 		// Restrict dials if limit is applied.
+		maxConcurrentDials := math.MaxInt
 		if flags.MaxDialIsActive() {
-			nodeCount = min(nodeCount, flags.Get().MaxConcurrentDials)
+			maxConcurrentDials = flags.Get().MaxConcurrentDials
 		}
-		nodes := enode.ReadNodes(iterator, nodeCount)
-		for _, node := range nodes {
-			info, _, err := convertToAddrInfo(node)
-			if err != nil {
-				continue
-			}
 
-			if info == nil {
-				continue
+		// Dial the peers in batches.
+		for start := 0; start < len(nodes); start += maxConcurrentDials {
+			stop := min(start+maxConcurrentDials, len(nodes))
+			for _, node := range nodes[start:stop] {
+				s.dialPeer(ctx, wg, node)
 			}
 
-			wg.Add(1)
-			go func() {
-				if err := s.connectWithPeer(ctx, *info); err != nil {
-					log.WithError(err).Tracef("Could not connect with peer %s", info.String())
-				}
-				wg.Done()
-			}()
+			// Wait for all dials to be completed.
+			wg.Wait()
 		}
-		// Wait for all dials to be completed.
-		wg.Wait()
+
+		_, missingPeerCountForTopic = peersSummary(topic, threshold)
 	}
+
+	log.WithField("topic", topic).Debug("Searching for new peers in the network - Success")
 	return true, nil
 }
 
@@ -183,11 +284,17 @@ func (s *Service) filterPeerForDataColumnsSubnet(index uint64) func(node *enode.
 // lower threshold to broadcast object compared to searching
 // for a subnet. So that even in the event of poor peer
 // connectivity, we can still broadcast an attestation.
-func (s *Service) hasPeerWithSubnet(topic string) bool {
+func (s *Service) hasPeerWithSubnet(subnetTopic string) bool {
 	// In the event peer threshold is lower, we will choose the lower
 	// threshold.
-	minPeers := mathutil.Min(1, uint64(flags.Get().MinimumPeersPerSubnet))
-	return len(s.pubsub.ListPeers(topic+s.Encoding().ProtocolSuffix())) >= int(minPeers) // lint:ignore uintcast -- Min peers can be safely cast to int.
+	minPeers := min(1, flags.Get().MinimumPeersPerSubnet)
+	topic := subnetTopic + s.Encoding().ProtocolSuffix()
+	peersWithSubnet := s.pubsub.ListPeers(topic)
+	peersWithSubnetCount := len(peersWithSubnet)
+
+	enoughPeers := peersWithSubnetCount >= minPeers
+
+	return enoughPeers
 }
 
 // Updates the service's discv5 listener record's attestation subnet