Use Connection Gater to Manage Peer Connections (#6243)

* use gater
* Merge branch 'master' into connGater
* Update beacon-chain/p2p/testing/p2p.go
* fmt
* gaz

beacon-chain/p2p/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "addr_factory.go",
         "broadcaster.go",
         "config.go",
+        "connection_gater.go",
         "dial_relay_node.go",
         "discovery.go",
         "doc.go",
@@ -63,8 +64,9 @@ go_library(
         "@com_github_libp2p_go_libp2p//:go_default_library",
         "@com_github_libp2p_go_libp2p//config:go_default_library",
         "@com_github_libp2p_go_libp2p//p2p/host/routed:go_default_library",
-        "@com_github_libp2p_go_libp2p_connmgr//:go_default_library",
         "@com_github_libp2p_go_libp2p_core//:go_default_library",
+        "@com_github_libp2p_go_libp2p_core//connmgr:go_default_library",
+        "@com_github_libp2p_go_libp2p_core//control:go_default_library",
         "@com_github_libp2p_go_libp2p_core//crypto:go_default_library",
         "@com_github_libp2p_go_libp2p_core//host:go_default_library",
         "@com_github_libp2p_go_libp2p_core//network:go_default_library",

beacon-chain/p2p/connection_gater.go

@@ -0,0 +1,41 @@
+package p2p
+
+import (
+	"github.com/libp2p/go-libp2p-core/control"
+	"github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multiaddr"
+	"github.com/sirupsen/logrus"
+)
+
+// InterceptPeerDial tests whether we're permitted to Dial the specified peer.
+func (s *Service) InterceptPeerDial(p peer.ID) (allow bool) {
+	return true
+}
+
+// InterceptAddrDial tests whether we're permitted to dial the specified
+// multiaddr for the given peer.
+func (s *Service) InterceptAddrDial(peer.ID, multiaddr.Multiaddr) (allow bool) {
+	return true
+}
+
+// InterceptAccept tests whether an incipient inbound connection is allowed.
+func (s *Service) InterceptAccept(n network.ConnMultiaddrs) (allow bool) {
+	if len(s.Peers().Active()) >= int(s.cfg.MaxPeers) {
+		log.WithFields(logrus.Fields{"peer": n.RemoteMultiaddr(),
+			"reason": "at peer limit"}).Trace("Not accepting inbound dial")
+		return false
+	}
+	return true
+}
+
+// InterceptSecured tests whether a given connection, now authenticated,
+// is allowed.
+func (s *Service) InterceptSecured(network.Direction, peer.ID, network.ConnMultiaddrs) (allow bool) {
+	return true
+}
+
+// InterceptUpgraded tests whether a fully capable connection is allowed.
+func (s *Service) InterceptUpgraded(network.Conn) (allow bool, reason control.DisconnectReason) {
+	return true, 0
+}

beacon-chain/p2p/discovery.go

@@ -135,7 +135,10 @@ func (s *Service) startDiscoveryV5(
 // 5) Peer's fork digest in their ENR matches that of
 // 	  our localnodes.
 func (s *Service) filterPeer(node *enode.Node) bool {
-	if len(s.Peers().Active()) >= int(s.cfg.MaxPeers) {
+	numOfConns := len(s.host.Network().Peers())
+	maxPeers := int(s.cfg.MaxPeers)
+	activePeers := len(s.Peers().Active())
+	if activePeers >= maxPeers || numOfConns >= maxPeers {
 		log.WithFields(logrus.Fields{"peer": node.String(),
 			"reason": "at peer limit"}).Trace("Not dialing peer")
 		return false

beacon-chain/p2p/interfaces.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/gogo/protobuf/proto"
+	"github.com/libp2p/go-libp2p-core/connmgr"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -41,6 +42,7 @@ type SetStreamHandler interface {
 type ConnectionHandler interface {
 	AddConnectionHandler(f func(ctx context.Context, id peer.ID) error, g func(context.Context, peer.ID) error)
 	AddDisconnectionHandler(f func(ctx context.Context, id peer.ID) error)
+	connmgr.ConnectionGater
 }
 
 // EncodingProvider provides p2p network encoding.

beacon-chain/p2p/options.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/libp2p/go-libp2p"
-	connmgr "github.com/libp2p/go-libp2p-connmgr"
 	noise "github.com/libp2p/go-libp2p-noise"
 	filter "github.com/multiformats/go-multiaddr"
 	ma "github.com/multiformats/go-multiaddr"
@@ -26,7 +25,8 @@ const (
 )
 
 // buildOptions for the libp2p host.
-func buildOptions(cfg *Config, ip net.IP, priKey *ecdsa.PrivateKey) []libp2p.Option {
+func (s *Service) buildOptions(ip net.IP, priKey *ecdsa.PrivateKey) []libp2p.Option {
+	cfg := s.cfg
 	listen, err := multiAddressBuilder(ip.String(), cfg.TCPPort)
 	if err != nil {
 		log.Fatalf("Failed to p2p listen: %v", err)
@@ -38,9 +38,7 @@ func buildOptions(cfg *Config, ip net.IP, priKey *ecdsa.PrivateKey) []libp2p.Opt
 		allowListSubnet(cfg.AllowListCIDR),
 		denyListSubnets(cfg.DenyListCIDR),
 		libp2p.UserAgent(version.GetBuildData()),
-		// Add one for the boot node and another for the relay, otherwise when we are close to maxPeers we will be above the high
-		// water mark and continually trigger pruning.
-		libp2p.ConnectionManager(connmgr.NewConnManager(int(cfg.MaxPeers), int(cfg.MaxPeers+peerBuffer), gracePeriod)),
+		libp2p.ConnectionGater(s),
 	}
 	if featureconfig.Get().EnableNoise {
 		// Enable NOISE for the beacon node

beacon-chain/p2p/service.go

@@ -129,8 +129,7 @@ func NewService(cfg *Config) (*Service, error) {
 		log.WithError(err).Error("Failed to create peer metadata")
 		return nil, err
 	}
-
-	opts := buildOptions(s.cfg, ipAddr, s.privKey)
+	opts := s.buildOptions(ipAddr, s.privKey)
 	h, err := libp2p.New(s.ctx, opts...)
 	if err != nil {
 		log.WithError(err).Error("Failed to create p2p host")

beacon-chain/p2p/testing/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
         "@com_github_gogo_protobuf//proto:go_default_library",
         "@com_github_libp2p_go_libp2p_blankhost//:go_default_library",
         "@com_github_libp2p_go_libp2p_core//:go_default_library",
+        "@com_github_libp2p_go_libp2p_core//control:go_default_library",
         "@com_github_libp2p_go_libp2p_core//host:go_default_library",
         "@com_github_libp2p_go_libp2p_core//network:go_default_library",
         "@com_github_libp2p_go_libp2p_core//peer:go_default_library",

beacon-chain/p2p/testing/p2p.go

@@ -13,12 +13,14 @@ import (
 	"github.com/gogo/protobuf/proto"
 	bhost "github.com/libp2p/go-libp2p-blankhost"
 	core "github.com/libp2p/go-libp2p-core"
+	"github.com/libp2p/go-libp2p-core/control"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
+	"github.com/multiformats/go-multiaddr"
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/beacon-chain/p2p/encoder"
 	"github.com/prysmaticlabs/prysm/beacon-chain/p2p/peers"
@@ -275,3 +277,28 @@ func (p *TestP2P) MetadataSeq() uint64 {
 func (p *TestP2P) AddPingMethod(reqFunc func(ctx context.Context, id peer.ID) error) {
 	// no-op
 }
+
+// InterceptPeerDial .
+func (p *TestP2P) InterceptPeerDial(peer.ID) (allow bool) {
+	return true
+}
+
+// InterceptAddrDial .
+func (p *TestP2P) InterceptAddrDial(peer.ID, multiaddr.Multiaddr) (allow bool) {
+	return true
+}
+
+// InterceptAccept .
+func (p *TestP2P) InterceptAccept(n network.ConnMultiaddrs) (allow bool) {
+	return true
+}
+
+// InterceptSecured .
+func (p *TestP2P) InterceptSecured(network.Direction, peer.ID, network.ConnMultiaddrs) (allow bool) {
+	return true
+}
+
+// InterceptUpgraded .
+func (p *TestP2P) InterceptUpgraded(network.Conn) (allow bool, reason control.DisconnectReason) {
+	return true, 0
+}