Use DHT to find peers

Author: jannikluhn

rolling-shutter/p2p/config.go

@@ -45,6 +45,7 @@ type Config struct {
 	ListenAddresses          []*address.P2PAddress
 	CustomBootstrapAddresses []*address.P2PAddress `comment:"Overwrite p2p boostrap nodes"`
 	Environment              env.Environment
+	DiscoveryNamespace       string `shconfig:",required" comment:"Must be unique for each instance id."`
 }
 
 func (c *Config) Name() string {
@@ -76,6 +77,7 @@ func (c *Config) SetExampleValues() error {
 		),
 	}
 	c.Environment = env.EnvironmentProduction
+	c.DiscoveryNamespace = "shutter-42"
 
 	p2pkey, err := keys.GenerateLibp2pPrivate(rand.Reader)
 	if err != nil {

rolling-shutter/p2p/dht.go

@@ -1,9 +1,17 @@
 package p2p
 
 import (
+	"context"
+	"time"
+
 	dht "github.com/libp2p/go-libp2p-kad-dht"
-	"github.com/libp2p/go-libp2p/core/peer"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/discovery"
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/libp2p/go-libp2p/p2p/discovery/util"
+	"github.com/rs/zerolog/log"
 
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/env"
 )
@@ -15,18 +23,24 @@ const (
 	dhtProtocolPrefix           protocol.ID = "/shutter"
 	dhtProtocolExtensionStaging protocol.ID = "/staging"
 	dhtProtocolExtensionLocal   protocol.ID = "/local"
+
+	findPeerInterval = 10 * time.Second
+)
+
+var (
+	peerLow    = pubsub.GossipSubDlo * 2
+	peerTarget = pubsub.GossipSubDhi * 3
+	peerHigh   = pubsub.GossipSubDhi * 6
 )
 
-func dhtRoutingOptions(
-	environment env.Environment,
-	bootstrapPeers ...peer.AddrInfo,
-) []dht.Option {
+func dhtRoutingOptions(config *p2pNodeConfig) []dht.Option {
 	// options with higher index in the array will overwrite existing ones
 	opts := []dht.Option{
 		dht.ProtocolPrefix(dhtProtocolPrefix),
+		dht.BootstrapPeers(config.BootstrapPeers...),
 	}
 
-	switch environment { //nolint: exhaustive
+	switch config.Environment { //nolint: exhaustive
 	case env.EnvironmentStaging:
 		opts = append(opts,
 			dht.ProtocolExtension(dhtProtocolExtensionStaging),
@@ -42,10 +56,56 @@ func dhtRoutingOptions(
 	default:
 	}
 
-	if len(bootstrapPeers) > 0 {
-		// this overwrites the option set before
-		opts = append(opts, dht.BootstrapPeers(bootstrapPeers...))
+	if config.IsBootstrapNode {
+		opts = append(opts, dht.Mode(dht.ModeServer))
 	}
 
 	return opts
 }
+
+func findPeers(ctx context.Context, h host.Host, d discovery.Discoverer, ns string) error {
+	log.Info().Str("namespace", ns).Msg("starting peer discovery")
+
+	ticker := time.NewTicker(findPeerInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			peersBefore := len(h.Network().Peers())
+			if peersBefore >= peerTarget {
+				continue
+			}
+
+			peers, err := util.FindPeers(ctx, d, ns)
+			if err != nil {
+				log.Error().Err(err).Msg("error finding peers")
+			}
+
+			newConnections := 0
+			failedDials := 0
+			for _, p := range peers {
+				if p.ID == h.ID() {
+					continue
+				}
+				if h.Network().Connectedness(p.ID) != network.Connected {
+					_, err = h.Network().DialPeer(ctx, p.ID)
+					if err != nil {
+						log.Error().Err(err).Str("peer", p.ID.String()).Msg("error dialing peer")
+						failedDials++
+					}
+					newConnections++
+				}
+			}
+
+			log.Debug().
+				Int("peers-before", peersBefore).
+				Int("peer-target", peerTarget).
+				Int("new-connections", newConnections).
+				Int("failed-dials", failedDials).
+				Msg("looking for peers")
+		}
+	}
+}

rolling-shutter/p2p/messaging.go

@@ -82,13 +82,10 @@ func New(config *Config) (*P2PMessaging, error) {
 		listenAddresses = append(listenAddresses, addr.Multiaddr)
 	}
 	cfg := &p2pNodeConfig{
-		ListenAddrs: listenAddresses,
-		PrivKey:     *config.P2PKey,
-		Environment: config.Environment,
-		// for now, disable those features, since
-		// they are not stable from our side
-		DisableTopicDHT:   true,
-		DisableRoutingDHT: true,
+		ListenAddrs:        listenAddresses,
+		PrivKey:            *config.P2PKey,
+		Environment:        config.Environment,
+		DiscoveryNamespace: config.DiscoveryNamespace,
 	}
 
 	bootstrapAddresses := config.CustomBootstrapAddresses

rolling-shutter/p2p/p2p.go

@@ -11,7 +11,9 @@ import (
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/p2p/discovery/routing"
+	"github.com/libp2p/go-libp2p/p2p/discovery/util"
 	rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
+	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"github.com/multiformats/go-multiaddr"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
@@ -52,20 +54,20 @@ type P2PNode struct {
 	mux         sync.Mutex
 	host        host.Host
 	dht         *dht.IpfsDHT
+	discovery   *routing.RoutingDiscovery
 	pubSub      *pubsub.PubSub
 	gossipRooms map[string]*gossipRoom
 
 	GossipMessages chan *pubsub.Message
 }
 
 type p2pNodeConfig struct {
-	ListenAddrs       []multiaddr.Multiaddr
-	BootstrapPeers    []peer.AddrInfo
-	PrivKey           keys.Libp2pPrivate
-	Environment       env.Environment
-	IsBootstrapNode   bool
-	DisableTopicDHT   bool
-	DisableRoutingDHT bool
+	ListenAddrs        []multiaddr.Multiaddr
+	BootstrapPeers     []peer.AddrInfo
+	PrivKey            keys.Libp2pPrivate
+	Environment        env.Environment
+	IsBootstrapNode    bool
+	DiscoveryNamespace string
 }
 
 func NewP2PNode(config p2pNodeConfig) *P2PNode {
@@ -118,6 +120,15 @@ func (p *P2PNode) Run(
 			return room.readLoop(ctx, p.GossipMessages)
 		})
 	}
+	runner.Go(func() error {
+		log.Info().Str("namespace", p.config.DiscoveryNamespace).Msg("starting advertizing discovery node")
+		util.Advertise(ctx, p.discovery, p.config.DiscoveryNamespace)
+		<-ctx.Done()
+		return ctx.Err()
+	})
+	runner.Go(func() error {
+		return findPeers(ctx, p.host, p.discovery, p.config.DiscoveryNamespace)
+	})
 	return nil
 }
 
@@ -141,18 +152,31 @@ func (p *P2PNode) init(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	p2pPubSub, err := createPubSub(ctx, p2pHost, p.config, hashTable)
+	discovery := routing.NewRoutingDiscovery(hashTable)
+	p2pPubSub, err := createPubSub(ctx, p2pHost, p.config, discovery)
 	if err != nil {
 		return err
 	}
 
 	p.host = p2pHost
 	p.dht = hashTable
+	p.discovery = discovery
 	p.pubSub = p2pPubSub
 	log.Info().Str("address", p.p2pAddress()).Msg("created libp2p host")
 	return nil
 }
 
+func createConnectionManager() (*connmgr.BasicConnMgr, error) {
+	// TODO: This starts a background goroutine. It works, but it's better to do that later in
+	// P2PNode.Run() when we have a proper context.
+	m, err := connmgr.NewConnManager(peerLow, peerHigh)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create connection manager")
+	}
+
+	return m, nil
+}
+
 func createHost(
 	ctx context.Context,
 	config p2pNodeConfig,
@@ -166,12 +190,15 @@ func createHost(
 	// This was a bug in the check function, reading the wrong config value to check against:
 	// https://github.com/libp2p/go-libp2p/issues/2628
 
+	connectionManager, err := createConnectionManager()
+	if err != nil {
+		return nil, nil, err
+	}
+
 	options := []libp2p.Option{
 		libp2p.Identity(&config.PrivKey.Key),
 		libp2p.ListenAddrs(config.ListenAddrs...),
-		// libp2p.DefaultTransports,
-		// libp2p.DefaultSecurity,
-		// libp2p.ConnectionManager(connectionManager),
+		libp2p.ConnectionManager(connectionManager),
 		libp2p.ProtocolVersion(protocolVersion),
 	}
 
@@ -192,11 +219,7 @@ func createHost(
 		return nil, nil, err
 	}
 
-	if config.DisableRoutingDHT {
-		return p2pHost, nil, err
-	}
-
-	opts := dhtRoutingOptions(config.Environment, config.BootstrapPeers...)
+	opts := dhtRoutingOptions(&config)
 	idht, err := dht.New(ctx, p2pHost, opts...)
 	if err != nil {
 		return nil, nil, err
@@ -212,7 +235,7 @@ func createPubSub(
 	ctx context.Context,
 	p2pHost host.Host,
 	config p2pNodeConfig,
-	hashTable *dht.IpfsDHT,
+	discovery *routing.RoutingDiscovery,
 ) (*pubsub.PubSub, error) {
 	gossipSubParams, peerScoreParams, peerScoreThresholds := makePubSubParams(pubSubParamsOptions{
 		isBootstrapNode: config.IsBootstrapNode,
@@ -222,14 +245,9 @@ func createPubSub(
 	pubsubOptions := []pubsub.Option{
 		pubsub.WithGossipSubParams(*gossipSubParams),
 		pubsub.WithPeerScore(peerScoreParams, peerScoreThresholds),
+		pubsub.WithDiscovery(discovery),
 	}
 
-	if !config.DisableTopicDHT {
-		pubsubOptions = append(
-			pubsubOptions,
-			pubsub.WithDiscovery(routing.NewRoutingDiscovery(hashTable)),
-		)
-	}
 	if config.IsBootstrapNode {
 		// enables the pubsub v1.1 feature to handle discovery and
 		// connection management over the PubSub protocol

rolling-shutter/p2p/params.go

@@ -17,6 +17,12 @@ func makePubSubParams(
 ) (*pubsub.GossipSubParams, *pubsub.PeerScoreParams, *pubsub.PeerScoreThresholds) {
 	gsDefault := pubsub.DefaultGossipSubParams()
 	gossipSubParams := &gsDefault
+
+	// modified defaults from ethereum consensus spec
+	// https://github.com/ethereum/consensus-specs/blob/5d80b1954a4b7a121aa36143d50b366727b66cbc/specs/phase0/p2p-interface.md#why-are-these-specific-gossip-parameters-chosen
+	gossipSubParams.HeartbeatInterval = 700 * time.Millisecond
+	gossipSubParams.HistoryLength = 6
+
 	// From the spec:
 	// to allow bootstrapping via PeerExchange (PX),
 	// the bootstrappers should not form a mesh, thus D=D_lo=D_hi=D_out=0