fix(dot/peerset): remove race conditions from peerset package

dot/network/connmgr_test.go

@@ -82,12 +82,12 @@ func TestMaxPeers(t *testing.T) {
 			continue
 		}
 
-		n.host.h.Peerstore().AddAddrs(ainfo.ID, ainfo.Addrs, peerstore.PermanentAddrTTL)
+		n.host.p2pHost.Peerstore().AddAddrs(ainfo.ID, ainfo.Addrs, peerstore.PermanentAddrTTL)
 		n.host.cm.peerSetHandler.AddPeer(0, ainfo.ID)
 	}
 
 	time.Sleep(200 * time.Millisecond)
-	p := nodes[0].host.h.Peerstore().Peers()
+	p := nodes[0].host.p2pHost.Peerstore().Peers()
 	require.LessOrEqual(t, max, len(p))
 }
 
@@ -152,15 +152,15 @@ func TestPersistentPeers(t *testing.T) {
 	time.Sleep(time.Millisecond * 600)
 
 	// B should have connected to A during bootstrap
-	conns := nodeB.host.h.Network().ConnsToPeer(nodeA.host.id())
+	conns := nodeB.host.p2pHost.Network().ConnsToPeer(nodeA.host.id())
 	require.NotEqual(t, 0, len(conns))
 
 	// if A disconnects from B, B should reconnect
 	nodeA.host.cm.peerSetHandler.DisconnectPeer(0, nodeB.host.id())
 
 	time.Sleep(time.Millisecond * 500)
 
-	conns = nodeB.host.h.Network().ConnsToPeer(nodeA.host.id())
+	conns = nodeB.host.p2pHost.Network().ConnsToPeer(nodeA.host.id())
 	require.NotEqual(t, 0, len(conns))
 }
 
@@ -239,7 +239,7 @@ func TestSetReservedPeer(t *testing.T) {
 
 	require.Equal(t, 2, node3.host.peerCount())
 
-	node3.host.h.Peerstore().AddAddrs(addrC.ID, addrC.Addrs, peerstore.PermanentAddrTTL)
+	node3.host.p2pHost.Peerstore().AddAddrs(addrC.ID, addrC.Addrs, peerstore.PermanentAddrTTL)
 	node3.host.cm.peerSetHandler.SetReservedPeer(0, addrC.ID)
 	time.Sleep(200 * time.Millisecond)
 

dot/network/discovery_test.go

@@ -35,7 +35,7 @@ func newTestDiscovery(t *testing.T, num int) []*discovery {
 		require.NoError(t, err)
 		disc := &discovery{
 			ctx: srvc.ctx,
-			h:   srvc.host.h,
+			h:   srvc.host.p2pHost,
 			ds:  ds,
 		}
 
@@ -200,7 +200,7 @@ func TestBeginDiscovery_ThreeNodes(t *testing.T) {
 	time.Sleep(time.Millisecond * 500)
 
 	// assert B and C can discover each other
-	addrs := nodeB.host.h.Peerstore().Addrs(nodeC.host.id())
+	addrs := nodeB.host.p2pHost.Peerstore().Addrs(nodeC.host.id())
 	require.NotEqual(t, 0, len(addrs))
 
 }

dot/network/host.go

@@ -68,7 +68,7 @@ const (
 // host wraps libp2p host with network host configuration and services
 type host struct {
 	ctx             context.Context
-	h               libp2phost.Host
+	p2pHost         libp2phost.Host
 	discovery       *discovery
 	bootnodes       []peer.AddrInfo
 	persistentPeers []peer.AddrInfo
@@ -211,7 +211,7 @@ func newHost(ctx context.Context, cfg *Config) (*host, error) {
 
 	host := &host{
 		ctx:             ctx,
-		h:               h,
+		p2pHost:         h,
 		discovery:       discovery,
 		bootnodes:       bns,
 		protocolID:      pid,
@@ -223,6 +223,7 @@ func newHost(ctx context.Context, cfg *Config) (*host, error) {
 	}
 
 	cm.host = host
+	cm.peerSetHandler.SetMessageProcessor(host)
 	return host, nil
 }
 
@@ -236,14 +237,14 @@ func (h *host) close() error {
 	}
 
 	// close libp2p host
-	err = h.h.Close()
+	err = h.p2pHost.Close()
 	if err != nil {
 		logger.Errorf("Failed to close libp2p host: %s", err)
 		return err
 	}
 
 	h.closeSync.Do(func() {
-		err = h.h.Peerstore().Close()
+		err = h.p2pHost.Peerstore().Close()
 		if err != nil {
 			logger.Errorf("Failed to close libp2p peerstore: %s", err)
 			return
@@ -260,28 +261,28 @@ func (h *host) close() error {
 
 // registerStreamHandler registers the stream handler for the given protocol id.
 func (h *host) registerStreamHandler(pid protocol.ID, handler func(libp2pnetwork.Stream)) {
-	h.h.SetStreamHandler(pid, handler)
+	h.p2pHost.SetStreamHandler(pid, handler)
 }
 
 // connect connects the host to a specific peer address
 func (h *host) connect(p peer.AddrInfo) (err error) {
-	h.h.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
+	h.p2pHost.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
 	ctx, cancel := context.WithTimeout(h.ctx, connectTimeout)
 	defer cancel()
-	err = h.h.Connect(ctx, p)
+	err = h.p2pHost.Connect(ctx, p)
 	return err
 }
 
 // bootstrap connects the host to the configured bootnodes
 func (h *host) bootstrap() {
 	for _, info := range h.persistentPeers {
-		h.h.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
+		h.p2pHost.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
 		h.cm.peerSetHandler.AddReservedPeer(0, info.ID)
 	}
 
 	for _, addrInfo := range h.bootnodes {
 		logger.Debugf("bootstrapping to peer %s", addrInfo.ID)
-		h.h.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, peerstore.PermanentAddrTTL)
+		h.p2pHost.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, peerstore.PermanentAddrTTL)
 		h.cm.peerSetHandler.AddPeer(0, addrInfo.ID)
 	}
 }
@@ -290,7 +291,7 @@ func (h *host) bootstrap() {
 // the newly created stream.
 func (h *host) send(p peer.ID, pid protocol.ID, msg Message) (libp2pnetwork.Stream, error) {
 	// open outbound stream with host protocol id
-	stream, err := h.h.NewStream(h.ctx, p, pid)
+	stream, err := h.p2pHost.NewStream(h.ctx, p, pid)
 	if err != nil {
 		logger.Tracef("failed to open new stream with peer %s using protocol %s: %s", p, pid, err)
 		return nil, err
@@ -334,12 +335,12 @@ func (h *host) writeToStream(s libp2pnetwork.Stream, msg Message) error {
 
 // id returns the host id
 func (h *host) id() peer.ID {
-	return h.h.ID()
+	return h.p2pHost.ID()
 }
 
 // Peers returns connected peers
 func (h *host) peers() []peer.ID {
-	return h.h.Network().Peers()
+	return h.p2pHost.Network().Peers()
 }
 
 // addReservedPeers adds the peers `addrs` to the protected peers list and connects to them
@@ -354,7 +355,7 @@ func (h *host) addReservedPeers(addrs ...string) error {
 		if err != nil {
 			return err
 		}
-		h.h.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, peerstore.PermanentAddrTTL)
+		h.p2pHost.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, peerstore.PermanentAddrTTL)
 		h.cm.peerSetHandler.AddReservedPeer(0, addrInfo.ID)
 	}
 
@@ -369,7 +370,7 @@ func (h *host) removeReservedPeers(ids ...string) error {
 			return err
 		}
 		h.cm.peerSetHandler.RemoveReservedPeer(0, peerID)
-		h.h.ConnManager().Unprotect(peerID, "")
+		h.p2pHost.ConnManager().Unprotect(peerID, "")
 	}
 
 	return nil
@@ -378,7 +379,7 @@ func (h *host) removeReservedPeers(ids ...string) error {
 // supportsProtocol checks if the protocol is supported by peerID
 // returns an error if could not get peer protocols
 func (h *host) supportsProtocol(peerID peer.ID, protocol protocol.ID) (bool, error) {
-	peerProtocols, err := h.h.Peerstore().SupportsProtocols(peerID, string(protocol))
+	peerProtocols, err := h.p2pHost.Peerstore().SupportsProtocols(peerID, string(protocol))
 	if err != nil {
 		return false, err
 	}
@@ -388,21 +389,21 @@ func (h *host) supportsProtocol(peerID peer.ID, protocol protocol.ID) (bool, err
 
 // peerCount returns the number of connected peers
 func (h *host) peerCount() int {
-	peers := h.h.Network().Peers()
+	peers := h.p2pHost.Network().Peers()
 	return len(peers)
 }
 
 // addrInfo returns the libp2p peer.AddrInfo of the host
 func (h *host) addrInfo() peer.AddrInfo {
 	return peer.AddrInfo{
-		ID:    h.h.ID(),
-		Addrs: h.h.Addrs(),
+		ID:    h.p2pHost.ID(),
+		Addrs: h.p2pHost.Addrs(),
 	}
 }
 
 // multiaddrs returns the multiaddresses of the host
 func (h *host) multiaddrs() (multiaddrs []ma.Multiaddr) {
-	addrs := h.h.Addrs()
+	addrs := h.p2pHost.Addrs()
 	for _, addr := range addrs {
 		multiaddr, err := ma.NewMultiaddr(fmt.Sprintf("%s/p2p/%s", addr, h.id()))
 		if err != nil {
@@ -415,16 +416,16 @@ func (h *host) multiaddrs() (multiaddrs []ma.Multiaddr) {
 
 // protocols returns all protocols currently supported by the node
 func (h *host) protocols() []string {
-	return h.h.Mux().Protocols()
+	return h.p2pHost.Mux().Protocols()
 }
 
 // closePeer closes connection with peer.
 func (h *host) closePeer(peer peer.ID) error {
-	return h.h.Network().ClosePeer(peer)
+	return h.p2pHost.Network().ClosePeer(peer)
 }
 
 func (h *host) closeProtocolStream(pID protocol.ID, p peer.ID) {
-	connToPeer := h.h.Network().ConnsToPeer(p)
+	connToPeer := h.p2pHost.Network().ConnsToPeer(p)
 	for _, c := range connToPeer {
 		for _, st := range c.GetStreams() {
 			if st.Protocol() != pID {
@@ -437,3 +438,38 @@ func (h *host) closeProtocolStream(pID protocol.ID, p peer.ID) {
 		}
 	}
 }
+
+// Process will connect, drop or reject a peer based on a peerset message
+func (h *host) Process(msg peerset.Message) {
+	peerID := msg.PeerID
+	if peerID == "" {
+		logger.Errorf("found empty peer id in peerset message")
+		return
+	}
+	switch msg.Status {
+	case peerset.Connect:
+		addrInfo := h.p2pHost.Peerstore().PeerInfo(peerID)
+		if len(addrInfo.Addrs) == 0 {
+			var err error
+			addrInfo, err = h.discovery.findPeer(peerID)
+			if err != nil {
+				logger.Warnf("failed to find peer id %s: %s", peerID, err)
+				return
+			}
+		}
+
+		err := h.connect(addrInfo)
+		if err != nil {
+			logger.Warnf("failed to open connection for peer %s: %s", peerID, err)
+			return
+		}
+		logger.Debugf("connection successful with peer %s", peerID)
+	case peerset.Drop, peerset.Reject:
+		err := h.closePeer(peerID)
+		if err != nil {
+			logger.Warnf("failed to close connection with peer %s: %s", peerID, err)
+			return
+		}
+		logger.Debugf("connection dropped successfully for peer %s", peerID)
+	}
+}

dot/network/host_test.go

@@ -170,13 +170,13 @@ func TestBootstrap(t *testing.T) {
 
 	peerCountA := nodeA.host.peerCount()
 	if peerCountA == 0 {
-		peerCountA := len(nodeA.host.h.Peerstore().Peers())
+		peerCountA := len(nodeA.host.p2pHost.Peerstore().Peers())
 		require.NotZero(t, peerCountA)
 	}
 
 	peerCountB := nodeB.host.peerCount()
 	if peerCountB == 0 {
-		peerCountB := len(nodeB.host.h.Peerstore().Peers())
+		peerCountB := len(nodeB.host.p2pHost.Peerstore().Peers())
 		require.NotZero(t, peerCountB)
 	}
 }
@@ -498,7 +498,7 @@ func Test_RemoveReservedPeers(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 
 	require.Equal(t, 1, nodeA.host.peerCount())
-	isProtected := nodeA.host.h.ConnManager().IsProtected(nodeB.host.addrInfo().ID, "")
+	isProtected := nodeA.host.p2pHost.ConnManager().IsProtected(nodeB.host.addrInfo().ID, "")
 	require.False(t, isProtected)
 
 	err = nodeA.host.removeReservedPeers("unknown_perr_id")
@@ -583,7 +583,7 @@ func TestPeerConnect(t *testing.T) {
 	nodeB.noGossip = true
 
 	addrInfoB := nodeB.host.addrInfo()
-	nodeA.host.h.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
+	nodeA.host.p2pHost.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
 	nodeA.host.cm.peerSetHandler.AddPeer(0, addrInfoB.ID)
 
 	time.Sleep(100 * time.Millisecond)
@@ -621,7 +621,7 @@ func TestBannedPeer(t *testing.T) {
 	nodeB.noGossip = true
 
 	addrInfoB := nodeB.host.addrInfo()
-	nodeA.host.h.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
+	nodeA.host.p2pHost.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
 	nodeA.host.cm.peerSetHandler.AddPeer(0, addrInfoB.ID)
 
 	time.Sleep(100 * time.Millisecond)
@@ -674,7 +674,7 @@ func TestPeerReputation(t *testing.T) {
 	nodeB.noGossip = true
 
 	addrInfoB := nodeB.host.addrInfo()
-	nodeA.host.h.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
+	nodeA.host.p2pHost.Peerstore().AddAddrs(addrInfoB.ID, addrInfoB.Addrs, peerstore.PermanentAddrTTL)
 	nodeA.host.cm.peerSetHandler.AddPeer(0, addrInfoB.ID)
 
 	time.Sleep(100 * time.Millisecond)

dot/network/light_test.go

@@ -113,7 +113,7 @@ func TestHandleLightMessage_Response(t *testing.T) {
 	}
 	require.NoError(t, err)
 
-	stream, err := s.host.h.NewStream(s.ctx, b.host.id(), s.host.protocolID+lightID)
+	stream, err := s.host.p2pHost.NewStream(s.ctx, b.host.id(), s.host.protocolID+lightID)
 	require.NoError(t, err)
 
 	// Testing empty request

dot/network/mdns.go

@@ -47,7 +47,7 @@ func (m *mdns) start() {
 	// create and start service
 	mdns, err := libp2pdiscovery.NewMdnsService(
 		m.host.ctx,
-		m.host.h,
+		m.host.p2pHost,
 		MDNSPeriod,
 		string(m.host.protocolID),
 	)
@@ -89,7 +89,7 @@ func (n Notifee) HandlePeerFound(p peer.AddrInfo) {
 		"Peer %s found using mDNS discovery, with host %s",
 		p.ID, n.host.id())
 
-	n.host.h.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
+	n.host.p2pHost.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
 	// connect to found peer
 	n.host.cm.peerSetHandler.AddPeer(0, p.ID)
 }

dot/network/mdns_test.go

@@ -42,13 +42,13 @@ func TestMDNS(t *testing.T) {
 
 	if peerCountA == 0 {
 		// check peerstore for disconnected peers
-		peerCountA := len(nodeA.host.h.Peerstore().Peers())
+		peerCountA := len(nodeA.host.p2pHost.Peerstore().Peers())
 		require.NotZero(t, peerCountA)
 	}
 
 	if peerCountB == 0 {
 		// check peerstore for disconnected peers
-		peerCountB := len(nodeB.host.h.Peerstore().Peers())
+		peerCountB := len(nodeB.host.p2pHost.Peerstore().Peers())
 		require.NotZero(t, peerCountB)
 	}
 }