Fetch data columns from multiple peers instead of just supernodes (#14977)

* Extract the block fetcher's peer selection logic for data columns so it can be used in both by range and by root requests

* Refactor data column sidecar request to send requests to multiple peers instead of supernodes

* Remove comment

* Remove unused method

* Add tests for dmissiblePeersForDataColumns

* Extract data column fetching into standalone functions

* Remove AdmissibleCustodyGroupsPeers and replace the final call with requests to multiple peers

* Apply suggestions from code review

Co-authored-by: Manu NALEPA <[email protected]>

* Wrap errors

* Use cached peedas.Info and properly convert custody groups to custody columns

* Rename filterPeersForRangeReq

* Preserve debugging descriptions when filtering out peers

* Remove unused functions.

* Initialize nested maps

* Fix comment

* First pass at retry logic for data column requests

* Select fresh peers for each retry

* Return an error if there are requested columns remaining

* Adjust errors

* Improve slightly the godoc.

* Improve wrapped error messages.

* `AdmissiblePeersForDataColumns`: Use value or `range`.

* Remove `convertCustodyGroupsToDataColumnsByPeer` since used only once.

* Minor fixes.

* Retry until we run out of peers

* Delete from the map of peers instead of filtering

* Remove unneeded break

* WIP: TestRequestDataColumnSidecars

* `RequestDataColumnSidecars`: Move the happy path in the for loop.

* Convert the peer ID to a node ID instead of using peer.EnodeID

* Extract AdmissiblePeersForDataColumns from a method into a function and use it (instead of a mock) in TestRequestDataColumnSidecars

* Track data column requests in tests to compare vs expectations

* Run gazelle

* Clean up test config changes so other tests don't break

* Clean up comments

* Minor changes.

* Add tests for peers that don't respond with all requested columns

* Respect MaxRequestDataColumnSidecars

---------

Co-authored-by: Manu NALEPA <[email protected]>
Co-authored-by: Manu NALEPA <[email protected]>

Author: niran

beacon-chain/p2p/custody.go

@@ -2,90 +2,13 @@ package p2p
 
 import (
 	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 )
 
-// AdmissibleCustodyGroupsPeers returns a list of peers that custody a super set of the local node's custody groups.
-func (s *Service) AdmissibleCustodyGroupsPeers(peers []peer.ID) ([]peer.ID, error) {
-	localCustodyGroupCount := s.cfg.CustodyInfo.ActualGroupCount()
-	return s.custodyGroupsAdmissiblePeers(peers, localCustodyGroupCount)
-}
-
-// AdmissibleCustodySamplingPeers returns a list of peers that custody a super set of the local node's sampling columns.
-func (s *Service) AdmissibleCustodySamplingPeers(peers []peer.ID) ([]peer.ID, error) {
-	localSubnetSamplingSize := s.cfg.CustodyInfo.CustodyGroupSamplingSize(peerdas.Actual)
-	return s.custodyGroupsAdmissiblePeers(peers, localSubnetSamplingSize)
-}
-
-// custodyGroupsAdmissiblePeers filters out `peers` that do not custody a super set of our own custody groups.
-func (s *Service) custodyGroupsAdmissiblePeers(peers []peer.ID, custodyGroupCount uint64) ([]peer.ID, error) {
-	// Get the total number of custody groups.
-	numberOfCustodyGroups := params.BeaconConfig().NumberOfCustodyGroups
-
-	// Retrieve the local node ID.
-	localNodeId := s.NodeID()
-
-	// Retrieve the local node info.
-	localNodeInfo, _, err := peerdas.Info(localNodeId, custodyGroupCount)
-	if err != nil {
-		return nil, errors.Wrap(err, "peer info")
-	}
-
-	// Retrieve the needed custody groups.
-	neededCustodyGroups := localNodeInfo.CustodyGroups
-
-	// Find the valid peers.
-	validPeers := make([]peer.ID, 0, len(peers))
-
-loop:
-	for _, pid := range peers {
-		// Get the custody group count of the remote peer.
-		remoteCustodyGroupCount := s.CustodyGroupCountFromPeer(pid)
-
-		// If the remote peer custodies less groups than we do, skip it.
-		if remoteCustodyGroupCount < custodyGroupCount {
-			continue
-		}
-
-		// Get the remote node ID from the peer ID.
-		remoteNodeID, err := ConvertPeerIDToNodeID(pid)
-		if err != nil {
-			return nil, errors.Wrap(err, "convert peer ID to node ID")
-		}
-
-		// Retrieve the remote peer info.
-		remotePeerInfo, _, err := peerdas.Info(remoteNodeID, remoteCustodyGroupCount)
-		if err != nil {
-			return nil, errors.Wrap(err, "peer info")
-		}
-
-		// Retrieve the custody groups of the remote peer.
-		remoteCustodyGroups := remotePeerInfo.CustodyGroups
-		remoteCustodyGroupsCount := uint64(len(remoteCustodyGroups))
-
-		// If the remote peers custodies all the possible columns, add it to the list.
-		if remoteCustodyGroupsCount == numberOfCustodyGroups {
-			validPeers = append(validPeers, pid)
-			continue
-		}
-
-		// Filter out invalid peers.
-		for custodyGroup := range neededCustodyGroups {
-			if !remoteCustodyGroups[custodyGroup] {
-				continue loop
-			}
-		}
-
-		// Add valid peer to list
-		validPeers = append(validPeers, pid)
-	}
-
-	return validPeers, nil
-}
+var _ DataColumnsHandler = (*Service)(nil)
 
 // custodyGroupCountFromPeerENR retrieves the custody count from the peer ENR.
 // If the ENR is not available, it defaults to the minimum number of custody groups

beacon-chain/p2p/custody_test.go

@@ -2,109 +2,20 @@ package p2p
 
 import (
 	"context"
-	"crypto/ecdsa"
-	"net"
 	"testing"
-	"time"
 
-	"github.com/ethereum/go-ethereum/p2p/discover"
-	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/enr"
-	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/network"
-	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/peers/scorers"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/wrapper"
-	ecdsaprysm "github.com/prysmaticlabs/prysm/v5/crypto/ecdsa"
-	prysmNetwork "github.com/prysmaticlabs/prysm/v5/network"
 	pb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1/metadata"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 )
 
-func createPeer(t *testing.T, privateKeyOffset int, custodyCount uint64) (*enr.Record, peer.ID, *ecdsa.PrivateKey) {
-	privateKeyBytes := make([]byte, 32)
-	for i := 0; i < 32; i++ {
-		privateKeyBytes[i] = byte(privateKeyOffset + i)
-	}
-
-	unmarshalledPrivateKey, err := crypto.UnmarshalSecp256k1PrivateKey(privateKeyBytes)
-	require.NoError(t, err)
-
-	privateKey, err := ecdsaprysm.ConvertFromInterfacePrivKey(unmarshalledPrivateKey)
-	require.NoError(t, err)
-
-	peerID, err := peer.IDFromPrivateKey(unmarshalledPrivateKey)
-	require.NoError(t, err)
-
-	record := &enr.Record{}
-	record.Set(peerdas.Cgc(custodyCount))
-	record.Set(enode.Secp256k1(privateKey.PublicKey))
-
-	return record, peerID, privateKey
-}
-
-func TestAdmissibleCustodyGroupsPeers(t *testing.T) {
-	genesisValidatorRoot := make([]byte, 32)
-
-	for i := 0; i < 32; i++ {
-		genesisValidatorRoot[i] = byte(i)
-	}
-
-	service := &Service{
-		cfg:                   &Config{CustodyInfo: &peerdas.CustodyInfo{}},
-		genesisTime:           time.Now(),
-		genesisValidatorsRoot: genesisValidatorRoot,
-		peers: peers.NewStatus(context.Background(), &peers.StatusConfig{
-			ScorerParams: &scorers.Config{},
-		}),
-	}
-
-	ipAddrString, err := prysmNetwork.ExternalIPv4()
-	require.NoError(t, err)
-	ipAddr := net.ParseIP(ipAddrString)
-
-	custodyRequirement := params.BeaconConfig().CustodyRequirement
-	dataColumnSidecarSubnetCount := params.BeaconConfig().DataColumnSidecarSubnetCount
-
-	// Peer 1 custodies exactly the same groups than us.
-	// (We use the same keys pair than ours for simplicity)
-	peer1Record, peer1ID, localPrivateKey := createPeer(t, 1, custodyRequirement)
-
-	// Peer 2 custodies all the groups.
-	peer2Record, peer2ID, _ := createPeer(t, 2, dataColumnSidecarSubnetCount)
-
-	// Peer 3 custodies different groups than us (but the same count).
-	// (We use the same public key than peer 2 for simplicity)
-	peer3Record, peer3ID, _ := createPeer(t, 3, custodyRequirement)
-
-	// Peer 4 custodies less groups than us.
-	peer4Record, peer4ID, _ := createPeer(t, 4, custodyRequirement-1)
-
-	createListener := func() (*discover.UDPv5, error) {
-		return service.createListener(ipAddr, localPrivateKey)
-	}
-
-	listener, err := newListener(createListener)
-	require.NoError(t, err)
-
-	service.dv5Listener = listener
-
-	service.peers.Add(peer1Record, peer1ID, nil, network.DirOutbound)
-	service.peers.Add(peer2Record, peer2ID, nil, network.DirOutbound)
-	service.peers.Add(peer3Record, peer3ID, nil, network.DirOutbound)
-	service.peers.Add(peer4Record, peer4ID, nil, network.DirOutbound)
-
-	actual, err := service.AdmissibleCustodyGroupsPeers([]peer.ID{peer1ID, peer2ID, peer3ID, peer4ID})
-	require.NoError(t, err)
-
-	expected := []peer.ID{peer1ID, peer2ID}
-	require.DeepSSZEqual(t, expected, actual)
-}
-
 func TestCustodyGroupCountFromPeer(t *testing.T) {
 	const (
 		expectedENR      uint64 = 7

beacon-chain/p2p/interfaces.go

@@ -115,6 +115,4 @@ type MetadataProvider interface {
 
 type DataColumnsHandler interface {
 	CustodyGroupCountFromPeer(peer.ID) uint64
-	AdmissibleCustodyGroupsPeers([]peer.ID) ([]peer.ID, error)
-	AdmissibleCustodySamplingPeers([]peer.ID) ([]peer.ID, error)
 }

beacon-chain/p2p/testing/fuzz_p2p.go

@@ -188,11 +188,3 @@ func (*FakeP2P) InterceptUpgraded(network.Conn) (allow bool, reason control.Disc
 func (*FakeP2P) CustodyGroupCountFromPeer(peer.ID) uint64 {
 	return 0
 }
-
-func (*FakeP2P) AdmissibleCustodyGroupsPeers(peers []peer.ID) ([]peer.ID, error) {
-	return peers, nil
-}
-
-func (*FakeP2P) AdmissibleCustodySamplingPeers(peers []peer.ID) ([]peer.ID, error) {
-	return peers, nil
-}

beacon-chain/p2p/testing/p2p.go

@@ -466,11 +466,3 @@ func (s *TestP2P) CustodyGroupCountFromPeer(pid peer.ID) uint64 {
 
 	return custodyGroupCount
 }
-
-func (*TestP2P) AdmissibleCustodyGroupsPeers(peers []peer.ID) ([]peer.ID, error) {
-	return peers, nil
-}
-
-func (*TestP2P) AdmissibleCustodySamplingPeers(peers []peer.ID) ([]peer.ID, error) {
-	return peers, nil
-}

beacon-chain/sync/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "block_batcher.go",
         "broadcast_bls_changes.go",
         "context.go",
+        "data_columns.go",
         "data_columns_reconstruct.go",
         "data_columns_sampling.go",
         "deadlines.go",
@@ -132,6 +133,7 @@ go_library(
         "//time:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
+        "@com_github_ethereum_go_ethereum//p2p/enode:go_default_library",
         "@com_github_hashicorp_golang_lru//:go_default_library",
         "@com_github_libp2p_go_libp2p//core:go_default_library",
         "@com_github_libp2p_go_libp2p//core/host:go_default_library",
@@ -164,6 +166,7 @@ go_test(
         "context_test.go",
         "data_columns_reconstruct_test.go",
         "data_columns_sampling_test.go",
+        "data_columns_test.go",
         "decode_pubsub_test.go",
         "error_test.go",
         "fork_watcher_test.go",
@@ -249,6 +252,7 @@ go_test(
         "//container/leaky-bucket:go_default_library",
         "//container/slice:go_default_library",
         "//crypto/bls:go_default_library",
+        "//crypto/ecdsa:go_default_library",
         "//crypto/rand:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//encoding/ssz/equality:go_default_library",
@@ -268,6 +272,7 @@ go_test(
         "@com_github_d4l3k_messagediff//:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_ethereum_go_ethereum//core/types:go_default_library",
+        "@com_github_ethereum_go_ethereum//p2p/enode:go_default_library",
         "@com_github_ethereum_go_ethereum//p2p/enr:go_default_library",
         "@com_github_golang_snappy//:go_default_library",
         "@com_github_libp2p_go_libp2p//:go_default_library",