remove the time.Sleep() dependency from dot/peerset/peerset_test.go

dot/network/helpers_test.go

@@ -90,7 +90,10 @@ func (s *testStreamHandler) readStream(stream libp2pnetwork.Stream,
 			return
 		} else if err != nil {
 			logger.Debugf("failed to read from stream using protocol %s: %s", stream.Protocol(), err)
-			_ = stream.Close()
+			err := stream.Close()
+			if err != nil {
+				logger.Warnf("failed to close stream: %s", err)
+			}
 			return
 		}
 
@@ -105,7 +108,10 @@ func (s *testStreamHandler) readStream(stream libp2pnetwork.Stream,
 		err = handler(stream, msg)
 		if err != nil {
 			logger.Errorf("failed to handle message %s from stream: %s", msg, err)
-			_ = stream.Close()
+			err := stream.Close()
+			if err != nil {
+				logger.Warnf("failed to close stream: %s", err)
+			}
 			return
 		}
 	}

dot/network/light.go

@@ -35,7 +35,10 @@ func (s *Service) decodeLightMessage(in []byte, peer peer.ID, _ bool) (Message,
 
 func (s *Service) handleLightMsg(stream libp2pnetwork.Stream, msg Message) (err error) {
 	defer func() {
-		_ = stream.Close()
+		err := stream.Close()
+		if err != nil {
+			logger.Warnf("failed to close stream: %s", err)
+		}
 	}()
 
 	lr, ok := msg.(*LightRequest)

dot/network/notifications.go

@@ -242,7 +242,10 @@ func closeOutboundStream(info *notificationsProtocol, peerID peer.ID, stream net
 	)
 
 	info.peersData.deleteOutboundHandshakeData(peerID)
-	_ = stream.Close()
+	err := stream.Close()
+	if err != nil {
+		logger.Warnf("failed to close outbound stream: %s", err)
+	}
 }
 
 func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtocol, msg NotificationsMessage) {

dot/network/notifications_test.go

@@ -269,7 +269,8 @@ func Test_HandshakeTimeout(t *testing.T) {
 	info.peersData.deleteOutboundHandshakeData(nodeB.host.id())
 	connAToB := nodeA.host.p2pHost.Network().ConnsToPeer(nodeB.host.id())
 	for _, stream := range connAToB[0].GetStreams() {
-		_ = stream.Close()
+		err := stream.Close()
+		require.NoError(t, err)
 	}
 
 	testHandshakeMsg := &BlockAnnounceHandshake{

dot/network/stream_manager.go

@@ -60,7 +60,10 @@ func (sm *streamManager) cleanupStreams() {
 		stream := data.stream
 
 		if time.Since(lastReceived) > cleanupStreamInterval {
-			_ = stream.Close()
+			err := stream.Close()
+			if err != nil {
+				logger.Warnf("failed to close inactive stream: %s", err)
+			}
 			delete(sm.streamData, id)
 		}
 	}

dot/network/sync.go

@@ -35,7 +35,10 @@ func (s *Service) DoBlockRequest(to peer.ID, req *BlockRequestMessage) (*BlockRe
 	}
 
 	defer func() {
-		_ = stream.Close()
+		err := stream.Close()
+		if err != nil {
+			logger.Warnf("failed to close stream: %s", err)
+		}
 	}()
 
 	if err = s.host.writeToStream(stream, req); err != nil {
@@ -102,7 +105,10 @@ func (s *Service) handleSyncMessage(stream libp2pnetwork.Stream, msg Message) er
 	}
 
 	defer func() {
-		_ = stream.Close()
+		err := stream.Close()
+		if err != nil {
+			logger.Warnf("failed to close stream: %s", err)
+		}
 	}()
 
 	if req, ok := msg.(*BlockRequestMessage); ok {

dot/peerset/peerset_test.go

@@ -4,6 +4,7 @@
 package peerset
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -78,9 +79,16 @@ func TestAddReservedPeers(t *testing.T) {
 
 	reservedPeers := peer.IDSlice{reservedPeer, reservedPeer2}
 
+	var waitGroup sync.WaitGroup
+
 	for _, peerID := range reservedPeers {
-		handler.AddReservedPeer(testSetID, peerID)
-		time.Sleep(time.Millisecond * 100)
+		waitGroup.Add(1)
+		go func() {
+			handler.AddReservedPeer(testSetID, peerID)
+			waitGroup.Done()
+		}()
+		waitGroup.Wait()
+		//time.Sleep(time.Millisecond * 100)
 
 		checkReservedNodePeerExists(t, ps, peerID)
 		checkPeerIsInNoSlotsNode(t, ps.peerState, peerID, testSetID)
@@ -152,14 +160,21 @@ func TestPeerSetIncoming(t *testing.T) {
 		},
 	}
 
+	var waitGroup sync.WaitGroup
+
 	for _, tt := range incomingPeers {
 
 		// all the incoming peers are unknow before calling the Incoming method
 		status := ps.peerState.peerStatus(testSetID, tt.pid)
 		require.Equal(t, unknownPeer, status)
 
-		handler.Incoming(testSetID, tt.pid)
-		time.Sleep(time.Millisecond * 100)
+		waitGroup.Add(1)
+		go func() {
+			handler.Incoming(testSetID, tt.pid)
+			waitGroup.Done()
+		}()
+		waitGroup.Wait()
+		//time.Sleep(time.Millisecond * 100)
 
 		checkNodePeerExists(t, ps.peerState, tt.pid)
 
@@ -190,11 +205,19 @@ func TestPeerSetDiscovered(t *testing.T) {
 	// reserved nodes should not increase the numOut count
 	checkPeerStateSetNumOut(t, ps.peerState, testSetID, 0)
 
-	handler.AddPeer(0, discovered1)
-	handler.AddPeer(0, discovered1)
-	handler.AddPeer(0, discovered2)
+	var waitGroup sync.WaitGroup
 
-	time.Sleep(200 * time.Millisecond)
+	waitGroup.Add(3)
+	go func() {
+		handler.AddPeer(0, discovered1)
+		handler.AddPeer(0, discovered1)
+		handler.AddPeer(0, discovered2)
+		waitGroup.Done()
+		waitGroup.Done()
+		waitGroup.Done()
+	}()
+	waitGroup.Wait()
+	//time.Sleep(200 * time.Millisecond)
 
 	checkNodePeerExists(t, ps.peerState, discovered1)
 	checkNodePeerExists(t, ps.peerState, discovered2)
@@ -227,9 +250,17 @@ func TestReAllocAfterBanned(t *testing.T) {
 
 	// We ban a node by setting its reputation under the threshold.
 	rep := newReputationChange(BannedThresholdValue-1, "")
-	handler.ReportPeer(rep, peer1)
 
-	time.Sleep(time.Millisecond * 100)
+	var waitGroup sync.WaitGroup
+
+	waitGroup.Add(1)
+	go func() {
+		handler.ReportPeer(rep, peer1)
+		waitGroup.Done()
+	}()
+	waitGroup.Wait()
+	//time.Sleep(time.Millisecond * 100)
+
 	checkMessageStatus(t, <-ps.resultMsgCh, Drop)
 
 	// banning a incoming peer should decrease the numIn count by 1
@@ -276,8 +307,15 @@ func TestRemovePeer(t *testing.T) {
 	require.Len(t, ps.peerState.nodes, 2)
 	checkPeerStateSetNumOut(t, ps.peerState, testSetID, 2)
 
-	handler.RemovePeer(testSetID, discovered1, discovered2)
-	time.Sleep(200 * time.Millisecond)
+	var waitGroup sync.WaitGroup
+
+	waitGroup.Add(1)
+	go func() {
+		handler.RemovePeer(testSetID, discovered1, discovered2)
+		waitGroup.Done()
+	}()
+	waitGroup.Wait()
+	//time.Sleep(200 * time.Millisecond)
 
 	require.Len(t, ps.resultMsgCh, 2)
 	for len(ps.resultMsgCh) != 0 {
@@ -304,9 +342,17 @@ func TestSetReservePeer(t *testing.T) {
 	require.Len(t, ps.reservedNode, 2)
 
 	newRsrPeerSet := peer.IDSlice{reservedPeer, peer.ID("newRsrPeer")}
-	// add newRsrPeer but remove reservedPeer2
-	handler.SetReservedPeer(testSetID, newRsrPeerSet...)
-	time.Sleep(200 * time.Millisecond)
+
+	var waitGroup sync.WaitGroup
+
+	waitGroup.Add(1)
+	go func() {
+		// add newRsrPeer but remove reservedPeer2
+		handler.SetReservedPeer(testSetID, newRsrPeerSet...)
+		waitGroup.Done()
+	}()
+	waitGroup.Wait()
+	//time.Sleep(200 * time.Millisecond)
 
 	checkPeerSetReservedNodeCount(t, ps, 2)
 	for _, p := range newRsrPeerSet {