prevent ConnManager from pruning peers excessively (#15681)

* prevent ConnManager from pruning peers excessively

* manu feedback

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>
Co-authored-by: james-prysm <90280386+james-prysm@users.noreply.github.com>

beacon-chain/p2p/BUILD.bazel

@@ -95,6 +95,7 @@ go_library(
         "@com_github_libp2p_go_libp2p//core/peer:go_default_library",
         "@com_github_libp2p_go_libp2p//core/peerstore:go_default_library",
         "@com_github_libp2p_go_libp2p//core/protocol:go_default_library",
+        "@com_github_libp2p_go_libp2p//p2p/net/connmgr:go_default_library",
         "@com_github_libp2p_go_libp2p//p2p/security/noise:go_default_library",
         "@com_github_libp2p_go_libp2p//p2p/transport/quic:go_default_library",
         "@com_github_libp2p_go_libp2p//p2p/transport/tcp:go_default_library",
@@ -184,6 +185,7 @@ go_test(
         "@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",
+        "@com_github_libp2p_go_libp2p//core/connmgr: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/config.go

@@ -11,6 +11,15 @@ import (
 
 // This is the default queue size used if we have specified an invalid one.
 const defaultPubsubQueueSize = 600
+const (
+	// defaultConnManagerPruneAbove sets the number of peers where ConnectionManager
+	// will begin to internally prune peers. This value is set based on the internal
+	// value of the libp2p DefaultConectionManager "high water mark". The "low water mark"
+	// is the number of peers where ConnManager will stop pruning. This value is computed
+	// by subtracting connManagerPruneAmount from the high water mark.
+	defaultConnManagerPruneAbove = 192
+	connManagerPruneAmount       = 32
+)
 
 // Config for the p2p service. These parameters are set from application level flags
 // to initialize the p2p service.
@@ -42,6 +51,18 @@ type Config struct {
 	ClockWaiter           startup.ClockWaiter
 }
 
+// connManagerLowHigh picks the low and high water marks for the connection manager based
+// on the MaxPeers setting. The high water mark will be at least the default high water mark
+// (192), or MaxPeers + 32, whichever is higher. The low water mark is set to be 32 less than
+// the high water mark. This is done to ensure the ConnManager never prunes peers that the
+// node has connected to based on the MaxPeers setting.
+func (cfg *Config) connManagerLowHigh() (int, int) {
+	maxPeersPlusMargin := int(cfg.MaxPeers) + connManagerPruneAmount
+	high := max(maxPeersPlusMargin, defaultConnManagerPruneAbove)
+	low := high - connManagerPruneAmount
+	return low, high
+}
+
 // validateConfig validates whether the values provided are accurate and will set
 // the appropriate values for those that are invalid.
 func validateConfig(cfg *Config) *Config {

beacon-chain/p2p/options.go

@@ -13,6 +13,7 @@ import (
 	mplex "github.com/libp2p/go-libp2p-mplex"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 	libp2pquic "github.com/libp2p/go-libp2p/p2p/transport/quic"
 	libp2ptcp "github.com/libp2p/go-libp2p/p2p/transport/tcp"
@@ -58,6 +59,22 @@ func MultiAddressBuilder(ip net.IP, tcpPort, quicPort uint) ([]ma.Multiaddr, err
 	return multiaddrs, nil
 }
 
+// setConnManagerOption sets the connection manager option for libp2p based on the
+// MaxPeers setting in the p2p config. If MaxPeers is set to a value higher than the
+// default high water mark, we create a new connection manager with a high water mark
+// that is higher than MaxPeers. Otherwise, we do not set a connection manager option
+// and allow the libp2p fallback defaults to be applied. Rationale below:
+// see: https://github.com/OffchainLabs/prysm/issues/15607
+func setConnManagerOption(cfg *Config, opts []libp2p.Option) ([]libp2p.Option, error) {
+	low, high := cfg.connManagerLowHigh()
+	cm, err := connmgr.NewConnManager(low, high)
+	if err != nil {
+		return nil, errors.Wrap(err, "new ConnManager")
+	}
+	opts = append(opts, libp2p.ConnectionManager(cm))
+	return opts, nil
+}
+
 // buildOptions for the libp2p host.
 func (s *Service) buildOptions(ip net.IP, priKey *ecdsa.PrivateKey) ([]libp2p.Option, error) {
 	cfg := s.cfg
@@ -84,7 +101,6 @@ func (s *Service) buildOptions(ip net.IP, priKey *ecdsa.PrivateKey) ([]libp2p.Op
 	if err != nil {
 		return nil, errors.Wrapf(err, "cannot get ID from public key: %s", ifaceKey.GetPublic().Type().String())
 	}
-
 	log.Infof("Running node with peer id of %s ", id.String())
 
 	options := []libp2p.Option{
@@ -98,6 +114,10 @@ func (s *Service) buildOptions(ip net.IP, priKey *ecdsa.PrivateKey) ([]libp2p.Op
 		libp2p.Security(noise.ID, noise.New),
 		libp2p.Ping(false), // Disable Ping Service.
 	}
+	options, err = setConnManagerOption(s.cfg, options)
+	if err != nil {
+		return nil, errors.Wrap(err, "set connection manager option")
+	}
 
 	if features.Get().EnableQUIC {
 		options = append(options, libp2p.Transport(libp2pquic.NewTransport))

beacon-chain/p2p/options_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -134,6 +135,59 @@ func TestDefaultMultiplexers(t *testing.T) {
 	assert.Equal(t, protocol.ID("/mplex/6.7.0"), cfg.Muxers[1].ID)
 }
 
+func TestSetConnManagerOption(t *testing.T) {
+	cases := []struct {
+		name      string
+		maxPeers  uint
+		highWater int
+	}{
+		{
+			name:      "MaxPeers lower than default high water mark",
+			maxPeers:  defaultConnManagerPruneAbove - 1,
+			highWater: defaultConnManagerPruneAbove,
+		},
+		{
+			name:      "MaxPeers equal to default high water mark",
+			maxPeers:  defaultConnManagerPruneAbove,
+			highWater: defaultConnManagerPruneAbove,
+		},
+		{
+			name:      "MaxPeers higher than default high water mark",
+			maxPeers:  defaultConnManagerPruneAbove + 1,
+			highWater: defaultConnManagerPruneAbove + 1 + connManagerPruneAmount,
+		},
+	}
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := &Config{MaxPeers: tt.maxPeers}
+			opts, err := setConnManagerOption(cfg, []libp2p.Option{})
+			assert.NoError(t, err)
+			_, high := cfg.connManagerLowHigh()
+			require.Equal(t, true, high > int(cfg.MaxPeers))
+
+			var libCfg libp2p.Config
+			require.NoError(t, libCfg.Apply(append(opts, libp2p.FallbackDefaults)...))
+			checkLimit(t, libCfg.ConnManager, high)
+		})
+	}
+}
+
+type connLimitGetter int
+
+func (m connLimitGetter) GetConnLimit() int {
+	return int(m)
+}
+
+// CheckLimit will return an error if the result of calling lg.GetConnLimit is greater than
+// the high water mark. So by checking the result of calling it with a value equal to and lower
+// than the expected value, we can determine the value it holds internally.
+func checkLimit(t *testing.T, cm connmgr.ConnManager, expected int) {
+	require.NoError(t, cm.CheckLimit(connLimitGetter(expected)), "Connection manager limit check failed")
+	if err := cm.CheckLimit(connLimitGetter(expected - 1)); err == nil {
+		t.Errorf("connection manager limit is below the expected value of %d", expected)
+	}
+}
+
 func TestMultiAddressBuilderWithID(t *testing.T) {
 	testCases := []struct {
 		name     string

changelog/kasey_fix-15607.md

@@ -0,0 +1,2 @@
+### Fixed
+- mitigate potential supernode clustering due to libp2p ConnManager pruning of non-supernodes, see https://github.com/OffchainLabs/prysm/issues/15607.