PeerDAS: move custody subnet count into helper function (#14117)

beacon-chain/blockchain/BUILD.bazel

@@ -69,7 +69,6 @@ go_library(
         "//beacon-chain/startup:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",

beacon-chain/blockchain/process_block.go

@@ -6,15 +6,14 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
-	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/blocks"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed"
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	coreTime "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/time"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/transition"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/das"
@@ -646,12 +645,8 @@ func (s *Service) isDataAvailableDataColumns(ctx context.Context, root [32]byte,
 	if len(kzgCommitments) == 0 {
 		return nil
 	}
-	custodiedSubnetCount := params.BeaconConfig().CustodyRequirement
-	if flags.Get().SubscribeToAllSubnets {
-		custodiedSubnetCount = params.BeaconConfig().DataColumnSidecarSubnetCount
-	}
 
-	colMap, err := peerdas.CustodyColumns(s.cfg.P2P.NodeID(), custodiedSubnetCount)
+	colMap, err := peerdas.CustodyColumns(s.cfg.P2P.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		return err
 	}

beacon-chain/core/peerdas/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas",
     visibility = ["//visibility:public"],
     deps = [
+        "//cmd/beacon-chain/flags:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/interfaces:go_default_library",

beacon-chain/core/peerdas/helpers.go

@@ -8,6 +8,8 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/holiman/uint256"
 	errors "github.com/pkg/errors"
+
+	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
@@ -300,3 +302,12 @@ func VerifyDataColumnSidecarKZGProofs(sc *ethpb.DataColumnSidecar) (bool, error)
 	}
 	return cKzg4844.VerifyCellKZGProofBatch(ckzgComms, rowIdx, colIdx, cells, proofs)
 }
+
+// CustodySubnetCount returns the number of subnets the node should participate in for custody.
+func CustodySubnetCount() uint64 {
+	count := params.BeaconConfig().CustodyRequirement
+	if flags.Get().SubscribeToAllSubnets {
+		count = params.BeaconConfig().DataColumnSidecarSubnetCount
+	}
+	return count
+}

beacon-chain/p2p/custody.go

@@ -6,16 +6,11 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
-	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 )
 
 func (s *Service) GetValidCustodyPeers(peers []peer.ID) ([]peer.ID, error) {
-	custodiedSubnetCount := params.BeaconConfig().CustodyRequirement
-	if flags.Get().SubscribeToAllSubnets {
-		custodiedSubnetCount = params.BeaconConfig().DataColumnSidecarSubnetCount
-	}
-	custodiedColumns, err := peerdas.CustodyColumns(s.NodeID(), custodiedSubnetCount)
+	custodiedColumns, err := peerdas.CustodyColumns(s.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		return nil, err
 	}

beacon-chain/p2p/discovery.go

@@ -17,6 +17,7 @@ import (
 	"github.com/prysmaticlabs/go-bitfield"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v5/config/features"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
@@ -275,11 +276,7 @@ func (s *Service) createLocalNode(
 	}
 
 	if features.Get().EnablePeerDAS {
-		custodySubnetEntry := CustodySubnetCount(params.BeaconConfig().CustodyRequirement)
-		if flags.Get().SubscribeToAllSubnets {
-			custodySubnetEntry = CustodySubnetCount(params.BeaconConfig().DataColumnSidecarSubnetCount)
-		}
-		localNode.Set(custodySubnetEntry)
+		localNode.Set(CustodySubnetCount(peerdas.CustodySubnetCount()))
 	}
 
 	localNode.SetFallbackIP(ipAddr)

beacon-chain/p2p/subnets.go

@@ -11,18 +11,18 @@ import (
 	"github.com/holiman/uint256"
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/go-bitfield"
+	"go.opencensus.io/trace"
+
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
+	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	"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"
-	"go.opencensus.io/trace"
-
-	"github.com/prysmaticlabs/prysm/v5/config/params"
 	pb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 )
 
@@ -220,7 +220,7 @@ func initializePersistentColumnSubnets(id enode.ID) error {
 	if ok && expTime.After(time.Now()) {
 		return nil
 	}
-	subsMap, err := peerdas.CustodyColumnSubnets(id, params.BeaconConfig().CustodyRequirement)
+	subsMap, err := peerdas.CustodyColumnSubnets(id, peerdas.CustodySubnetCount())
 	if err != nil {
 		return err
 	}

beacon-chain/sync/data_columns_reconstruct.go

@@ -8,14 +8,14 @@ import (
 
 	cKzg4844 "github.com/ethereum/c-kzg-4844/bindings/go"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
-	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
-	"github.com/sirupsen/logrus"
 )
 
 const broadCastMissingDataColumnsTimeIntoSlot = 3 * time.Second
@@ -111,12 +111,7 @@ func (s *Service) reconstructDataColumns(ctx context.Context, verifiedRODataColu
 	}
 
 	// Retrieve the custodied columns.
-	custodiedSubnetCount := params.BeaconConfig().CustodyRequirement
-	if flags.Get().SubscribeToAllSubnets {
-		custodiedSubnetCount = params.BeaconConfig().DataColumnSidecarSubnetCount
-	}
-
-	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), custodiedSubnetCount)
+	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		return errors.Wrap(err, "custodied columns")
 	}
@@ -210,12 +205,7 @@ func (s *Service) scheduleReconstructedDataColumnsBroadcast(
 		}
 
 		// Get the data columns we should store.
-		custodiedSubnetCount := params.BeaconConfig().CustodyRequirement
-		if flags.Get().SubscribeToAllSubnets {
-			custodiedSubnetCount = params.BeaconConfig().DataColumnSidecarSubnetCount
-		}
-
-		custodiedDataColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), custodiedSubnetCount)
+		custodiedDataColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), peerdas.CustodySubnetCount())
 		if err != nil {
 			log.WithError(err).Error("Custody columns")
 		}

beacon-chain/sync/initial-sync/BUILD.bazel

@@ -34,7 +34,6 @@ go_library(
         "//beacon-chain/verification:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
-        "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/interfaces:go_default_library",

beacon-chain/sync/initial-sync/blocks_fetcher.go

@@ -10,6 +10,9 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"go.opencensus.io/trace"
+
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filesystem"
@@ -31,8 +34,6 @@ import (
 	p2ppb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
-	"github.com/sirupsen/logrus"
-	"go.opencensus.io/trace"
 )
 
 const (
@@ -575,8 +576,9 @@ func populateBlockWithColumns(bw blocks2.BlockWithROBlobs, columns []blocks.RODa
 		return bw, errDidntPopulate
 	}
 	colsPersub := params.BeaconConfig().NumberOfColumns / params.BeaconConfig().DataColumnSidecarSubnetCount
-	if len(columns) != int(params.BeaconConfig().CustodyRequirement*colsPersub) {
-		return bw, errors.Errorf("unequal custodied columns provided, got %d instead of %d", len(columns), int(params.BeaconConfig().CustodyRequirement))
+	subnetCount := peerdas.CustodySubnetCount()
+	if len(columns) != int(subnetCount*colsPersub) {
+		return bw, errors.Errorf("unequal custodied columns provided, got %d instead of %d", len(columns), subnetCount)
 	}
 	for ci := range columns {
 		if err := verify.ColumnAlignsWithBlock(columns[ci], blk); err != nil {
@@ -655,7 +657,7 @@ func (f *blocksFetcher) fetchColumnsFromPeer(ctx context.Context, bwb []blocks2.
 		return bwb, nil
 	}
 	// Construct request message based on required custodied columns.
-	custodyCols, err := peerdas.CustodyColumns(f.p2p.NodeID(), params.BeaconConfig().CustodyRequirement)
+	custodyCols, err := peerdas.CustodyColumns(f.p2p.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		return nil, err
 	}
@@ -679,7 +681,8 @@ func (f *blocksFetcher) fetchColumnsFromPeer(ctx context.Context, bwb []blocks2.
 		if err != nil {
 			return nil, err
 		}
-		remoteCustody, err := peerdas.CustodyColumns(nid, params.BeaconConfig().CustodyRequirement)
+
+		remoteCustody, err := peerdas.CustodyColumns(nid, f.p2p.CustodyCountFromRemotePeer(p))
 		if err != nil {
 			return nil, err
 		}

beacon-chain/sync/initial-sync/service.go

@@ -28,7 +28,6 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v5/config/features"
-	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v5/crypto/rand"
@@ -343,17 +342,11 @@ func (s *Service) missingColumnRequest(roBlock blocks.ROBlock, store *filesystem
 		return nil, errors.Wrapf(err, "error checking existing blobs for checkpoint sync block root %#x", blockRoot)
 	}
 
-	// Get the number of columns we should custody.
-	custodyRequirement := params.BeaconConfig().CustodyRequirement
-	if features.Get().EnablePeerDAS {
-		custodyRequirement = fieldparams.NumberOfColumns
-	}
-
 	// Get our node ID.
 	nodeID := s.cfg.P2P.NodeID()
 
 	// Get the custodied columns.
-	custodiedColumns, err := peerdas.CustodyColumns(nodeID, custodyRequirement)
+	custodiedColumns, err := peerdas.CustodyColumns(nodeID, peerdas.CustodySubnetCount())
 	if err != nil {
 		return nil, errors.Wrap(err, "custody columns")
 	}

beacon-chain/sync/rpc_beacon_blocks_by_root.go

@@ -13,7 +13,6 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/types"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/sync/verify"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
-	"github.com/prysmaticlabs/prysm/v5/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v5/config/features"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
@@ -271,14 +270,8 @@ func (s *Service) constructPendingColumnRequest(root [32]byte) (types.DataColumn
 		return nil, errors.Wrap(err, "column indices")
 	}
 
-	// Compute how many subnets we should custody.
-	custodiedSubnetCount := params.BeaconConfig().CustodyRequirement
-	if flags.Get().SubscribeToAllSubnets {
-		custodiedSubnetCount = params.BeaconConfig().DataColumnSidecarSubnetCount
-	}
-
 	// Retrieve the columns we should custody.
-	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), custodiedSubnetCount)
+	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		return nil, errors.Wrap(err, "custody columns")
 	}

beacon-chain/sync/rpc_data_column_sidecars_by_root.go

@@ -9,6 +9,9 @@ import (
 
 	libp2pcore "github.com/libp2p/go-libp2p/core"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"go.opencensus.io/trace"
+
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/peerdas"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filesystem"
@@ -21,8 +24,6 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
-	"github.com/sirupsen/logrus"
-	"go.opencensus.io/trace"
 )
 
 func (s *Service) dataColumnSidecarByRootRPCHandler(ctx context.Context, msg interface{}, stream libp2pcore.Stream) error {
@@ -71,14 +72,8 @@ func (s *Service) dataColumnSidecarByRootRPCHandler(ctx context.Context, msg int
 		return errors.Wrapf(err, "unexpected error computing min valid blob request slot, current_slot=%d", cs)
 	}
 
-	// Compute all custodied subnets.
-	custodiedSubnets := params.BeaconConfig().CustodyRequirement
-	if flags.Get().SubscribeToAllSubnets {
-		custodiedSubnets = params.BeaconConfig().DataColumnSidecarSubnetCount
-	}
-
 	// Compute all custodied columns.
-	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), custodiedSubnets)
+	custodiedColumns, err := peerdas.CustodyColumns(s.cfg.p2p.NodeID(), peerdas.CustodySubnetCount())
 	if err != nil {
 		log.WithError(err).Errorf("unexpected error retrieving the node id")
 		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)