rotation of outbound peers (#3037)

To make the p2p network uniformly random, nodes need to be able to
change their peers periodically (otherwise network would be centralized
at bootstrap peers). This PR implements an algorithm based on stable
hashing, which makes nodes assign random priority to every node ID, and
connect to peers with higher priority.

This pr also separates inbound and outbound connection pools to simplify
logic. In particular it is possible that there will be 2 concurrent
connections between 2 peers (inbound + outbound). Avoiding duplicate
connections is best effort - nodes try not to dial peers that they are
connected to, but in case 2 peers dial each other at the same time they
let those 2 connections be.

Basic requirements:
* peermanager maintains a pex table: addresses of peers of our peers.
This bounds the network view size (even though it is an unauthenticated
network) that our node needs to maintain and provides enough exposure to
select new peers. Peers are periodically reporting their connections,
therefore peermanager keeps only fresh addresses
* on startup a spike of dials is expected - node will try to connect to
peers that it was connected to before restart
* during stable operation node will dial peers at a low rate like 1/s or
0.1/s, and the node should be selected from a fresh set of addresses -
i.e. we cannot snapshot a list of currently available addresses and try
to dial them all (it will take hours)
* despite low dial rate, node should attempt to round robin over the
ever changing peer candidates set - i.e. it should not get stuck dialing
the same bad address over and over
* in the stable-hash-based approach, each peer ID obtains a priority for
dialing - it should be taken into account
* implementation should support replacing low priority peers with higher
priority peers (to support convergence to a random graph). The churn of
the connections should be low though, so that connection efficiency is
not affected. My initial guesstimate would be that we should allow
replacing a connection every ~1min.
* We need to support a pex table with ~100 * 100 = 10k addresses total
(100 connections per peer is a safe estimate with the current
implementation). Whether we can affort just rank all the addresses on
every dial attempt is a borderline IMO.
* the addresses inserted to pex table should be made available for
dialing ASAP, without any active polling, if possible.

Author: pompon0

sei-tendermint/config/config.go

@@ -652,7 +652,7 @@ type P2PConfig struct {
 	// MaxOutboundConnections limits the number of outbound connections to regular (non-persistent) peers.
 	// It should be significantly lower than MaxConnections, unless
 	// the node is supposed to have a small number of connections altogether.
-	MaxOutboundConnections uint
+	MaxOutboundConnections *uint `mapstructure:"max-outbound-connections"`
 
 	// MaxIncomingConnectionAttempts rate limits the number of incoming connection
 	// attempts per IP address.
@@ -703,14 +703,10 @@ type P2PConfig struct {
 // DefaultP2PConfig returns a default configuration for the peer-to-peer layer
 func DefaultP2PConfig() *P2PConfig {
 	return &P2PConfig{
-		ListenAddress:   "tcp://127.0.0.1:26656",
-		ExternalAddress: "",
-		UPNP:            false,
-		MaxConnections:  100,
-		// TODO(gprusak): decrease to 10, once PEX is improved to:
-		// * exchange both inbound and outbound connections information
-		// * exchange information on handshake as well.
-		MaxOutboundConnections:        100,
+		ListenAddress:                 "tcp://127.0.0.1:26656",
+		ExternalAddress:               "",
+		UPNP:                          false,
+		MaxConnections:                100,
 		MaxIncomingConnectionAttempts: 100,
 		FlushThrottleTimeout:          100 * time.Millisecond,
 		MaxPacketMsgPayloadSize:       1000000,

sei-tendermint/internal/p2p/address.go

@@ -35,21 +35,6 @@ type NodeAddress struct {
 	Port     uint16
 }
 
-var cgnat = netip.MustParsePrefix("100.64.0.0/10")
-
-// IsPublic checks if the address is routable from the public internet.
-// It is good enough to exclude internal addresses of cloud providers.
-// As a simplification, it treats non-IP Hostnames (DNS addresses) as public.
-// TODO(gprusak): DNS addresses should be eliminated from PEX entirely - all
-// addresses should be resolved locally and only then advertised to peers.
-func (a NodeAddress) IsPublic() bool {
-	ip, err := netip.ParseAddr(a.Hostname)
-	if err != nil {
-		return true
-	}
-	return ip.IsGlobalUnicast() && !ip.IsPrivate() && !cgnat.Contains(ip.Unmap())
-}
-
 // ParseNodeAddress parses a node address URL into a NodeAddress, normalizing
 // and validating it.
 func ParseNodeAddress(urlString string) (NodeAddress, error) {

sei-tendermint/internal/p2p/address_test.go

@@ -5,7 +5,6 @@ import (
 	"testing"
 
 	"github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519"
-	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/tcp"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/types"
@@ -278,21 +277,6 @@ func TestNodeAddress_String(t *testing.T) {
 	}
 }
 
-func TestNodeAddress_IsPublic(t *testing.T) {
-	rng := utils.TestRng()
-	id := makeNodeID(rng)
-	testcases := map[string]bool{
-		"192.168.1.10":  false,
-		"93.184.216.34": true,
-		"example.com":   true,
-		"100.64.0.1":    false,
-	}
-	for hostname, isPublic := range testcases {
-		addr := NodeAddress{NodeID: id, Hostname: hostname, Port: defaultPort}
-		require.Equal(t, isPublic, addr.IsPublic())
-	}
-}
-
 func TestNodeAddress_Validate(t *testing.T) {
 	id := types.NodeID("00112233445566778899aabbccddeeff00112233")
 	testcases := []struct {

sei-tendermint/internal/p2p/channel.go

@@ -79,12 +79,12 @@ func (ch *Channel[T]) send(msg T, queues ...*Queue[sendMsg]) {
 }
 
 func (ch *Channel[T]) Send(msg T, to types.NodeID) {
-	c, ok := ch.router.peerManager.Conns().Get(to)
+	c, ok := GetAny(ch.router.peerManager.Conns(), to)
 	if !ok {
 		logger.Debug("dropping message for unconnected peer", "peer", to, "channel", ch.desc.ID)
 		return
 	}
-	if _, contains := c.peerChannels[ch.desc.ID]; !contains {
+	if _, contains := c.Channels[ch.desc.ID]; !contains {
 		// reactor tried to send a message across a channel that the
 		// peer doesn't have available. This is a known issue due to
 		// how peer subscriptions work:
@@ -98,7 +98,7 @@ func (ch *Channel[T]) Send(msg T, to types.NodeID) {
 func (ch *Channel[T]) Broadcast(msg T) {
 	var queues []*Queue[sendMsg]
 	for _, c := range ch.router.peerManager.Conns().All() {
-		if _, ok := c.peerChannels[ch.desc.ID]; ok {
+		if _, ok := c.Channels[ch.desc.ID]; ok {
 			queues = append(queues, c.sendQueue)
 		}
 	}