fix another data race by backporting fb3ceba

WARNING: DATA RACE
Read at 0x00c005b2f898 by goroutine 50785:
  gitlab.com/thorchain/tss/go-tss/p2p.(*StreamMgr).ReleaseStream()
      /home/alex/workspace/github.com/zeta-chain/go-tss/p2p/stream_helper.go:53 +0x1f0
  gitlab.com/thorchain/tss/go-tss/p2p.(*Communication).ReleaseStream()
      /home/alex/workspace/github.com/zeta-chain/go-tss/p2p/communication.go:484 +0x104
  gitlab.com/thorchain/tss/go-tss/tss.(*TssServer).KeySign.func1()
      /home/alex/workspace/github.com/zeta-chain/go-tss/tss/keysign.go:219 +0xc4
  runtime.deferreturn()
      /usr/local/go/src/runtime/panic.go:602 +0x5c
  gitlab.com/thorchain/tss/go-tss/tss.(*FourNodeScaleZetaSuite).doTestConcurrentKeySign.func1()
      /home/alex/workspace/github.com/zeta-chain/go-tss/tss/tss_4nodes_zeta_test.go:198 +0x294
  gitlab.com/thorchain/tss/go-tss/tss.(*FourNodeScaleZetaSuite).doTestConcurrentKeySign.gowrap1()
      /home/alex/workspace/github.com/zeta-chain/go-tss/tss/tss_4nodes_zeta_test.go:203 +0x44

Previous write at 0x00c005b2f898 by goroutine 72246:
  gitlab.com/thorchain/tss/go-tss/p2p.(*StreamMgr).AddStream()
      /home/alex/workspace/github.com/zeta-chain/go-tss/p2p/stream_helper.go:76 +0x13c
  gitlab.com/thorchain/tss/go-tss/p2p.(*Communication).readFromStream()
      /home/alex/workspace/github.com/zeta-chain/go-tss/p2p/communication.go:176 +0x334
  gitlab.com/thorchain/tss/go-tss/p2p.(*Communication).handleStream()
      /home/alex/workspace/github.com/zeta-chain/go-tss/p2p/communication.go:195 +0x10c
  gitlab.com/thorchain/tss/go-tss/p2p.(*Communication).handleStream-fm()
      <autogenerated>:1 +0x44
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).SetStreamHandler.func1()
      /home/alex/go/pkg/mod/github.com/zeta-chain/[email protected]/p2p/host/basic/basic_host.go:580 +0xb4
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler.gowrap1()
      /home/alex/go/pkg/mod/github.com/zeta-chain/[email protected]/p2p/host/basic/basic_host.go:421 +0x6c

p2p/communication.go

@@ -173,13 +173,14 @@ func (c *Communication) readFromStream(stream network.Stream) {
 			return
 		}
 		c.logger.Debug().Msgf(">>>>>>>[%s] %s", wrappedMsg.MessageType, string(wrappedMsg.Payload))
-		c.streamMgr.AddStream(wrappedMsg.MsgID, stream)
 		channel := c.getSubscriber(wrappedMsg.MessageType, wrappedMsg.MsgID)
 		if nil == channel {
 			c.logger.Debug().Msgf("no MsgID %s found for this message", wrappedMsg.MsgID)
 			c.logger.Debug().Msgf("no MsgID %s found for this message", wrappedMsg.MessageType)
+			_ = stream.Close()
 			return
 		}
+		c.streamMgr.AddStream(wrappedMsg.MsgID, stream)
 		channel <- &Message{
 			PeerID:  stream.Conn().RemotePeer(),
 			Payload: dataBuf,

p2p/party_coordinator.go

@@ -66,16 +66,16 @@ func (pc *PartyCoordinator) Stop() {
 }
 
 func (pc *PartyCoordinator) processRespMsg(respMsg *messages.JoinPartyLeaderComm, stream network.Stream) {
-	pc.streamMgr.AddStream(respMsg.ID, stream)
-
 	remotePeer := stream.Conn().RemotePeer()
 	pc.joinPartyGroupLock.Lock()
 	peerGroup, ok := pc.peersGroup[respMsg.ID]
 	pc.joinPartyGroupLock.Unlock()
 	if !ok {
 		pc.logger.Info().Msgf("message ID from peer(%s) can not be found", remotePeer)
+		_ = stream.Close()
 		return
 	}
+	pc.streamMgr.AddStream(respMsg.ID, stream)
 	if remotePeer == peerGroup.getLeader() {
 		peerGroup.setLeaderResponse(respMsg)
 		peerGroup.notify <- true

p2p/stream_helper.go

@@ -58,6 +58,7 @@ func (sm *StreamMgr) ReleaseStream(msgID string) {
 		}
 		sm.streamLocker.Lock()
 		delete(sm.unusedStreams, msgID)
+		delete(sm.unusedStreams, "UNKNOWN")
 		sm.streamLocker.Unlock()
 	}
 }