refactor(network): refactoring peer manager (#787)

network/dht.go

@@ -13,7 +13,6 @@ type dhtService struct {
 	ctx      context.Context
 	host     lp2phost.Host
 	kademlia *lp2pdht.IpfsDHT
-	peerMgr  *peerMgr
 	logger   *logger.SubLogger
 }
 
@@ -24,45 +23,32 @@ func newDHTService(ctx context.Context, host lp2phost.Host, protocolID lp2pcore.
 	if conf.Bootstrapper {
 		mode = lp2pdht.ModeServer
 	}
-	bootsrapAddrs := PeerAddrsToAddrInfo(conf.BootstrapAddrs)
+	bootstrapAddrs := PeerAddrsToAddrInfo(conf.BootstrapAddrs)
 	opts := []lp2pdht.Option{
 		lp2pdht.Mode(mode),
 		lp2pdht.ProtocolPrefix(protocolID),
-		lp2pdht.BootstrapPeers(bootsrapAddrs...),
+		lp2pdht.BootstrapPeers(bootstrapAddrs...),
 	}
 
 	kademlia, err := lp2pdht.New(ctx, host, opts...)
 	if err != nil {
-		logger.Panic("unable to start DHT service", "error", err)
-		return nil
+		panic(err)
 	}
 
-	err = kademlia.Bootstrap(ctx)
-	if err != nil {
-		panic(err.Error())
-	}
-
-	bootstrap := newPeerMgr(ctx, host, host.Network(), kademlia,
-		bootsrapAddrs, conf.MinConns, conf.MaxConns, logger)
-
 	return &dhtService{
 		ctx:      ctx,
 		host:     host,
 		kademlia: kademlia,
-		peerMgr:  bootstrap,
 		logger:   logger,
 	}
 }
 
-func (dht *dhtService) Start() error {
-	dht.peerMgr.Start()
-	return nil
+func (s *dhtService) Start() error {
+	return s.kademlia.Bootstrap(s.ctx)
 }
 
-func (dht *dhtService) Stop() {
-	if err := dht.kademlia.Close(); err != nil {
-		dht.logger.Error("unable to close Kademlia", "error", err)
+func (s *dhtService) Stop() {
+	if err := s.kademlia.Close(); err != nil {
+		s.logger.Error("unable to close Kademlia", "error", err)
 	}
-
-	dht.peerMgr.Stop()
 }

network/mdns.go

@@ -34,30 +34,30 @@ func newMdnsService(ctx context.Context, host lp2phost.Host, logger *logger.SubL
 // HandlePeerFound connects to peers discovered via mDNS. Once they're connected,
 // the PubSub system will automatically start interacting with them if they also
 // support PubSub.
-func (mdns *mdnsService) HandlePeerFound(pi lp2ppeer.AddrInfo) {
-	ctx, cancel := context.WithTimeout(mdns.ctx, time.Second*10)
+func (s *mdnsService) HandlePeerFound(pi lp2ppeer.AddrInfo) {
+	ctx, cancel := context.WithTimeout(s.ctx, time.Second*10)
 	defer cancel()
 
-	if pi.ID != mdns.host.ID() {
-		mdns.logger.Debug("connecting to new peer", "addr", pi.Addrs, "id", pi.ID.Pretty())
-		if err := mdns.host.Connect(ctx, pi); err != nil {
-			mdns.logger.Error("error on connecting to peer", "id", pi.ID.Pretty(), "error", err)
+	if pi.ID != s.host.ID() {
+		s.logger.Debug("connecting to new peer", "addr", pi.Addrs, "id", pi.ID.Pretty())
+		if err := s.host.Connect(ctx, pi); err != nil {
+			s.logger.Error("error on connecting to peer", "id", pi.ID.Pretty(), "error", err)
 		}
 	}
 }
 
-func (mdns *mdnsService) Start() error {
-	err := mdns.service.Start()
+func (s *mdnsService) Start() error {
+	err := s.service.Start()
 	if err != nil {
 		return LibP2PError{Err: err}
 	}
 
 	return nil
 }
 
-func (mdns *mdnsService) Stop() {
-	err := mdns.service.Close()
+func (s *mdnsService) Stop() {
+	err := s.service.Close()
 	if err != nil {
-		mdns.logger.Error("unable to close the network", "error", err)
+		s.logger.Error("unable to close the network", "error", err)
 	}
 }

network/mdns_test.go

@@ -33,4 +33,7 @@ func TestMDNS(t *testing.T) {
 	se := shouldReceiveEvent(t, net2, EventTypeStream).(*StreamMessage)
 	assert.Equal(t, se.Source, net1.SelfID())
 	assert.Equal(t, readData(t, se.Reader, len(msg)), msg)
+
+	net1.Stop()
+	net2.Stop()
 }

network/network.go

@@ -35,6 +35,7 @@ type network struct {
 	host           lp2phost.Host
 	mdns           *mdnsService
 	dht            *dhtService
+	peerMgr        *peerMgr
 	stream         *streamService
 	gossip         *gossipService
 	notifee        *NotifeeService
@@ -152,6 +153,7 @@ func newNetwork(networkName string, conf *Config, opts []lp2p.Option) (*network,
 		if err != nil {
 			return nil, LibP2PError{Err: err}
 		}
+
 		opts = append(opts,
 			lp2p.EnableRelay(),
 			lp2p.EnableAutoRelayWithStaticRelays(static),
@@ -189,6 +191,7 @@ func newNetwork(networkName string, conf *Config, opts []lp2p.Option) (*network,
 	streamProtocolID := lp2pcore.ProtocolID(fmt.Sprintf("/%s/stream/v1", n.name))
 
 	n.dht = newDHTService(n.ctx, n.host, kadProtocolID, conf, n.logger)
+	n.peerMgr = newPeerMgr(ctx, host, n.dht.kademlia, conf, n.logger)
 	n.stream = newStreamService(ctx, n.host, streamProtocolID, relayAddrs, n.eventChannel, n.logger)
 	n.gossip = newGossipService(ctx, n.host, n.eventChannel, conf, n.logger)
 	n.notifee = newNotifeeService(n.host, n.eventChannel, n.logger, streamProtocolID)
@@ -217,8 +220,9 @@ func (n *network) Start() error {
 	}
 	n.gossip.Start()
 	n.stream.Start()
+	n.peerMgr.Start()
 
-	n.logger.Info("network started", "addr", n.host.Addrs())
+	n.logger.Info("network started", "addr", n.host.Addrs(), "id", n.host.ID())
 	return nil
 }
 
@@ -232,6 +236,7 @@ func (n *network) Stop() {
 	n.dht.Stop()
 	n.gossip.Stop()
 	n.stream.Stop()
+	n.peerMgr.Stop()
 
 	if err := n.host.Close(); err != nil {
 		n.logger.Error("unable to close the network", "error", err)

network/network_test.go

@@ -150,21 +150,21 @@ func TestNetwork(t *testing.T) {
 		relayAddrs = append(relayAddrs, addr2)
 	}
 
+	bootstrapPort := ts.RandInt32(9999) + 10000
+	publicPort := ts.RandInt32(9999) + 10000
+
 	// Bootstrap node
 	confB := testConfig()
-	bootstrapPort := ts.RandInt32(9999) + 10000
 	confB.Bootstrapper = true
 	confB.Listens = []string{
 		fmt.Sprintf("/ip4/127.0.0.1/tcp/%v", bootstrapPort),
-		fmt.Sprintf("/ip6/::1/tcp/%v", bootstrapPort),
 	}
 	fmt.Println("Starting Bootstrap node")
 	networkB := makeTestNetwork(t, confB, []lp2p.Option{
 		lp2p.ForceReachabilityPublic(),
 	})
 	bootstrapAddresses := []string{
 		fmt.Sprintf("/ip4/127.0.0.1/tcp/%v/p2p/%v", bootstrapPort, networkB.SelfID().String()),
-		fmt.Sprintf("/ip6/::1/tcp/%v/p2p/%v", bootstrapPort, networkB.SelfID().String()),
 	}
 	assert.NoError(t, networkB.JoinConsensusTopic())
 
@@ -173,14 +173,14 @@ func TestNetwork(t *testing.T) {
 	confP.EnableNAT = true
 	confP.BootstrapAddrs = bootstrapAddresses
 	confP.Listens = []string{
-		"/ip4/127.0.0.1/tcp/0",
-		"/ip6/::1/tcp/0",
+		fmt.Sprintf("/ip4/127.0.0.1/tcp/%v", publicPort),
 	}
 	fmt.Println("Starting Public node")
 	networkP := makeTestNetwork(t, confP, []lp2p.Option{
 		lp2p.ForceReachabilityPublic(),
 	})
 	assert.NoError(t, networkP.JoinConsensusTopic())
+	publicAddrInfo, _ := MakeAddressInfo(fmt.Sprintf("/ip4/127.0.0.1/tcp/%v/p2p/%s", publicPort, networkP.SelfID()))
 
 	// Private node M
 	confM := testConfig()
@@ -189,7 +189,6 @@ func TestNetwork(t *testing.T) {
 	confM.BootstrapAddrs = bootstrapAddresses
 	confM.Listens = []string{
 		"/ip4/127.0.0.1/tcp/0",
-		"/ip6/::1/tcp/0",
 	}
 	fmt.Println("Starting Private node M")
 	networkM := makeTestNetwork(t, confM, []lp2p.Option{
@@ -204,26 +203,31 @@ func TestNetwork(t *testing.T) {
 	confN.BootstrapAddrs = bootstrapAddresses
 	confN.Listens = []string{
 		"/ip4/127.0.0.1/tcp/0",
-		"/ip6/::1/tcp/0",
 	}
 	fmt.Println("Starting Private node N")
 	networkN := makeTestNetwork(t, confN, []lp2p.Option{
 		lp2p.ForceReachabilityPrivate(),
 	})
 	assert.NoError(t, networkN.JoinConsensusTopic())
 
-	// Private node X, doesn't join consensus topic
+	// Private node X, doesn't join consensus topic and without relay address
 	confX := testConfig()
 	confX.EnableRelay = false
 	confX.BootstrapAddrs = bootstrapAddresses
 	confX.Listens = []string{
 		"/ip4/127.0.0.1/tcp/0",
-		"/ip6/::1/tcp/0",
 	}
 	fmt.Println("Starting Private node X")
 	networkX := makeTestNetwork(t, confX, []lp2p.Option{
 		lp2p.ForceReachabilityPrivate(),
 	})
+
+	assert.NoError(t, networkB.Start())
+	assert.NoError(t, networkP.Start())
+	assert.NoError(t, networkM.Start())
+	assert.NoError(t, networkN.Start())
+	assert.NoError(t, networkX.Start())
+
 	time.Sleep(2 * time.Second)
 
 	t.Run("all nodes have at least one connection to the bootstrap node B", func(t *testing.T) {
@@ -275,12 +279,25 @@ func TestNetwork(t *testing.T) {
 	})
 
 	t.Run("node P (public) is directly accessible by nodes M and N (private behind NAT)", func(t *testing.T) {
+		require.NoError(t, networkM.host.Connect(networkM.ctx, *publicAddrInfo))
+
 		msgM := []byte("test-stream-from-m")
 
 		require.NoError(t, networkM.SendTo(msgM, networkP.SelfID()))
-		eB := shouldReceiveEvent(t, networkP, EventTypeStream).(*StreamMessage)
-		assert.Equal(t, eB.Source, networkM.SelfID())
-		assert.Equal(t, readData(t, eB.Reader, len(msgM)), msgM)
+		eP := shouldReceiveEvent(t, networkP, EventTypeStream).(*StreamMessage)
+		assert.Equal(t, eP.Source, networkM.SelfID())
+		assert.Equal(t, readData(t, eP.Reader, len(msgM)), msgM)
+	})
+
+	t.Run("node P (public) is directly accessible by node X (private behind NAT, without relay)", func(t *testing.T) {
+		require.NoError(t, networkX.host.Connect(networkX.ctx, *publicAddrInfo))
+
+		msgX := []byte("test-stream-from-x")
+
+		require.NoError(t, networkX.SendTo(msgX, networkP.SelfID()))
+		eP := shouldReceiveEvent(t, networkP, EventTypeStream).(*StreamMessage)
+		assert.Equal(t, eP.Source, networkX.SelfID())
+		assert.Equal(t, readData(t, eP.Reader, len(msgX)), msgX)
 	})
 
 	t.Run("node P (public) is directly accessible by node B (bootstrap)", func(t *testing.T) {

network/notifee.go

@@ -42,13 +42,11 @@ func (n *NotifeeService) Connected(lp2pn lp2pnetwork.Network, conn lp2pnetwork.C
 
 			protocols, _ := lp2pn.Peerstore().SupportsProtocols(peerID, n.protocolID)
 			if len(protocols) > 0 {
-				n.eventChannel <- &ConnectEvent{PeerID: peerID}
-				return
+				break
 			}
 		}
 
-		n.logger.Info("unable to get supported protocols", "pid", peerID)
-		_ = n.host.Network().ClosePeer(peerID)
+		n.eventChannel <- &ConnectEvent{PeerID: peerID}
 	}()
 }
 

network/peermgr.go

@@ -34,16 +34,17 @@ type peerMgr struct {
 // newPeerMgr creates a new Peer Manager instance.
 // Peer Manager attempts to establish connections with other nodes when the
 // number of connections falls below the minimum threshold.
-func newPeerMgr(ctx context.Context, h lp2phost.Host, d lp2pnet.Dialer, dht *lp2pdht.IpfsDHT,
-	bootstrapAddrs []lp2ppeer.AddrInfo, minConns int, maxConns int, logger *logger.SubLogger,
+func newPeerMgr(ctx context.Context, h lp2phost.Host, dht *lp2pdht.IpfsDHT,
+	conf *Config, logger *logger.SubLogger,
 ) *peerMgr {
+	bootstrapAddrs := PeerAddrsToAddrInfo(conf.BootstrapAddrs)
 	b := &peerMgr{
 		ctx:            ctx,
 		bootstrapAddrs: bootstrapAddrs,
-		minConns:       minConns,
-		maxConns:       maxConns,
+		minConns:       conf.MinConns,
+		maxConns:       conf.MaxConns,
 		host:           h,
-		dialer:         d,
+		dialer:         h.Network(),
 		dht:            dht,
 		logger:         logger,
 	}
@@ -119,12 +120,5 @@ func (mgr *peerMgr) checkConnectivity() {
 
 			ConnectAsync(mgr.ctx, mgr.host, pi, mgr.logger)
 		}
-
-		mgr.logger.Debug("expanding the connections")
-
-		err := mgr.dht.Bootstrap(mgr.ctx)
-		if err != nil {
-			mgr.logger.Warn("peer discovery may suffer", "error", err)
-		}
 	}
 }