fix(dot/network): fix data race and add t.Parallel to all tests

dot/network/config.go

@@ -105,7 +105,9 @@ type Config struct {
 	SlotDuration time.Duration
 
 	Telemetry telemetry.Client
-	Metrics   metrics.IntervalConfig
+
+	MessageCacheTTL time.Duration
+	Metrics         metrics.IntervalConfig
 }
 
 // build checks the configuration, sets up the private key for the network service,

dot/network/helpers_test.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"io"
 	"math/big"
+	"sync"
 	"testing"
 
 	"github.com/ChainSafe/gossamer/dot/types"
@@ -18,15 +19,19 @@ import (
 )
 
 type testStreamHandler struct {
+	sync.Mutex
+
 	messages map[peer.ID][]Message
 	decoder  messageDecoder
-	exit     bool
+
+	eofCh chan struct{}
 }
 
 func newTestStreamHandler(decoder messageDecoder) *testStreamHandler {
 	return &testStreamHandler{
 		messages: make(map[peer.ID][]Message),
 		decoder:  decoder,
+		eofCh:    make(chan struct{}),
 	}
 }
 
@@ -42,9 +47,11 @@ func (s *testStreamHandler) handleStream(stream libp2pnetwork.Stream) {
 }
 
 func (s *testStreamHandler) handleMessage(stream libp2pnetwork.Stream, msg Message) error {
+
+	s.Lock()
+	defer s.Unlock()
 	msgs := s.messages[stream.Conn().RemotePeer()]
 	s.messages[stream.Conn().RemotePeer()] = append(msgs, msg)
-
 	announceHandshake := &BlockAnnounceHandshake{
 		BestBlockNumber: 0,
 	}
@@ -69,13 +76,10 @@ func (s *testStreamHandler) readStream(stream libp2pnetwork.Stream,
 	peer peer.ID, decoder messageDecoder, handler messageHandler) {
 	msgBytes := make([]byte, maxBlockResponseSize)
 
-	defer func() {
-		s.exit = true
-	}()
-
 	for {
 		tot, err := readStream(stream, msgBytes)
 		if errors.Is(err, io.EOF) {
+			s.eofCh <- struct{}{}
 			return
 		} else if err != nil {
 			logger.Debugf("failed to read from stream using protocol %s: %s", stream.Protocol(), err)
@@ -100,6 +104,14 @@ func (s *testStreamHandler) readStream(stream libp2pnetwork.Stream,
 	}
 }
 
+func (s *testStreamHandler) messagesFrom(peerID peer.ID) (messages []Message, ok bool) {
+	s.Lock()
+	defer s.Unlock()
+
+	messages, ok = s.messages[peerID]
+	return messages, ok
+}
+
 var starting, _ = variadic.NewUint64OrHash(uint64(1))
 
 var one = uint32(1)

dot/network/host.go

@@ -201,7 +201,7 @@ func newHost(ctx context.Context, cfg *Config) (*host, error) {
 			return int64(1)
 		},
 	}
-	msgCache, err := newMessageCache(config, msgCacheTTL)
+	msgCache, err := newMessageCache(config, cfg.MessageCacheTTL)
 	if err != nil {
 		return nil, err
 	}

dot/network/host_test.go

@@ -4,7 +4,7 @@
 package network
 
 import (
-	"fmt"
+	"regexp"
 	"testing"
 	"time"
 
@@ -50,11 +50,10 @@ func mustNewMultiAddr(s string) (a ma.Multiaddr) {
 func TestExternalAddrsPublicIP(t *testing.T) {
 	t.Parallel()
 
-	port := availablePort(t)
 	config := &Config{
 		BasePath:    t.TempDir(),
 		PublicIP:    "10.0.5.2",
-		Port:        port,
+		Port:        availablePort(t),
 		NoBootstrap: true,
 		NoMDNS:      true,
 	}
@@ -65,6 +64,8 @@ func TestExternalAddrsPublicIP(t *testing.T) {
 	privateIPs, err := newPrivateIPFilters()
 	require.NoError(t, err)
 
+	multiAddrRegex := regexp.MustCompile("^/ip4/(127.0.0.1|10.0.5.2)/tcp/[0-9]+$")
+
 	for i, addr := range addrInfo.Addrs {
 		switch i {
 		case len(addrInfo.Addrs) - 1:
@@ -73,16 +74,14 @@ func TestExternalAddrsPublicIP(t *testing.T) {
 		default:
 			require.False(t, privateIPs.AddrBlocked(addr))
 		}
-	}
 
-	expected := []ma.Multiaddr{
-		mustNewMultiAddr(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", port)),
-		mustNewMultiAddr(fmt.Sprintf("/ip4/10.0.5.2/tcp/%d", port)),
+		require.True(t, multiAddrRegex.MatchString(addr.String()))
 	}
-	assert.Equal(t, addrInfo.Addrs, expected)
 }
 
 func TestExternalAddrsPublicDNS(t *testing.T) {
+	t.Parallel()
+
 	config := &Config{
 		BasePath:    t.TempDir(),
 		PublicDNS:   "alice",
@@ -99,7 +98,6 @@ func TestExternalAddrsPublicDNS(t *testing.T) {
 		mustNewMultiAddr("/dns/alice/tcp/7001"),
 	}
 	assert.Equal(t, addrInfo.Addrs, expected)
-
 }
 
 // test host connect method
@@ -222,7 +220,7 @@ func TestSend(t *testing.T) {
 
 	time.Sleep(TestMessageTimeout)
 
-	msg, ok := handler.messages[nodeA.host.id()]
+	msg, ok := handler.messagesFrom(nodeA.host.id())
 	require.True(t, ok)
 	require.Equal(t, 1, len(msg))
 	require.Equal(t, testBlockReqMessage, msg[0])
@@ -273,24 +271,32 @@ func TestExistingStream(t *testing.T) {
 	require.NoError(t, err)
 
 	time.Sleep(TestMessageTimeout)
-	require.NotNil(t, handlerB.messages[nodeA.host.id()], "node B timeout waiting for message from node A")
+
+	messages, ok := handlerB.messagesFrom(nodeA.host.id())
+	require.True(t, ok, "node B timed out waiting for message from node A")
+	assert.Len(t, messages, 1)
 
 	// node A uses the stream to send a second message
 	err = nodeA.host.writeToStream(stream, testBlockReqMessage)
 	require.NoError(t, err)
-	require.NotNil(t, handlerB.messages[nodeA.host.id()], "node B timeout waiting for message from node A")
+
+	_, ok = handlerB.messagesFrom(nodeA.host.id())
+	require.True(t, ok, "node B timed out waiting for message from node A")
 
 	// node B opens the stream to send the first message
 	stream, err = nodeB.host.send(addrInfoA.ID, nodeB.host.protocolID, testBlockReqMessage)
 	require.NoError(t, err)
 
 	time.Sleep(TestMessageTimeout)
-	require.NotNil(t, handlerA.messages[nodeB.host.id()], "node A timeout waiting for message from node B")
+	_, ok = handlerA.messagesFrom(nodeB.host.id())
+	require.True(t, ok, "node A timed out waiting for message from node B")
 
 	// node B uses the stream to send a second message
 	err = nodeB.host.writeToStream(stream, testBlockReqMessage)
 	require.NoError(t, err)
-	require.NotNil(t, handlerA.messages[nodeB.host.id()], "node A timeout waiting for message from node B")
+
+	_, ok = handlerA.messagesFrom(nodeB.host.id())
+	require.True(t, ok, "node A timed out waiting for message from node B")
 }
 
 func TestStreamCloseMetadataCleanup(t *testing.T) {
@@ -531,9 +537,9 @@ func TestStreamCloseEOF(t *testing.T) {
 
 	nodeB := createTestService(t, configB)
 	nodeB.noGossip = true
+
 	handler := newTestStreamHandler(testBlockRequestMessageDecoder)
 	nodeB.host.registerStreamHandler(nodeB.host.protocolID, handler.handleStream)
-	require.False(t, handler.exit)
 
 	addrInfoB := nodeB.host.addrInfo()
 	err := nodeA.host.connect(addrInfoB)
@@ -548,14 +554,20 @@ func TestStreamCloseEOF(t *testing.T) {
 
 	stream, err := nodeA.host.send(addrInfoB.ID, nodeB.host.protocolID, testBlockReqMessage)
 	require.NoError(t, err)
-	require.False(t, handler.exit)
 
 	err = stream.Close()
 	require.NoError(t, err)
 
-	time.Sleep(TestBackoffTimeout)
+	timeout := time.NewTimer(TestBackoffTimeout)
 
-	require.True(t, handler.exit)
+	select {
+	case <-timeout.C:
+		t.Fatal("stream handler does not exit after stream closed")
+	case <-handler.eofCh:
+		if !timeout.Stop() {
+			<-timeout.C
+		}
+	}
 }
 
 // Test to check the nodes connection by peer set manager
@@ -678,6 +690,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.cm.peerSetHandler.AddPeer(0, addrInfoB.ID)
 
@@ -694,6 +707,9 @@ func TestPeerReputation(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 
 	rep, err := nodeA.host.cm.peerSetHandler.PeerReputation(addrInfoB.ID)
+
+	const zeroReputation int32 = 0
+
 	require.NoError(t, err)
-	require.Greater(t, rep, int32(0))
+	require.Greater(t, rep, zeroReputation)
 }

dot/network/notifications.go

@@ -264,6 +264,7 @@ func closeOutboundStream(info *notificationsProtocol, peerID peer.ID, stream lib
 	)
 
 	info.outboundHandshakeData.Delete(peerID)
+	fmt.Println(">>>>>> closed outbound stream")
 	_ = stream.Close()
 }
 
@@ -289,12 +290,12 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
 	}
 
 	if s.host.messageCache != nil && s.host.messageCache.exists(peer, msg) {
-		logger.Tracef("message has already been sent, ignoring: peer=%s msg=%s", peer, msg)
+		logger.Debugf("message has already been sent, ignoring: peer=%s msg=%s", peer, msg)
 		return
 	}
 
 	// we've completed the handshake with the peer, send message directly
-	logger.Tracef("sending message to peer %s using protocol %s: %s", peer, info.protocolID, msg)
+	logger.Debugf("sending message to peer %s using protocol %s: %s", peer, info.protocolID, msg)
 	if err := s.host.writeToStream(stream, msg); err != nil {
 		logger.Debugf("failed to send message to peer %s: %s", peer, err)
 
@@ -310,7 +311,7 @@ func (s *Service) sendData(peer peer.ID, hs Handshake, info *notificationsProtoc
 		}
 	}
 
-	logger.Tracef("successfully sent message on protocol %s to peer %s: message=", info.protocolID, peer, msg)
+	logger.Debugf("successfully sent message on protocol %s to peer %s: message=", info.protocolID, peer, msg)
 	s.host.cm.peerSetHandler.ReportPeer(peerset.ReputationChange{
 		Value:  peerset.GossipSuccessValue,
 		Reason: peerset.GossipSuccessReason,