p2p: do not register closing peer (algorand#6086)

Co-authored-by: cce <[email protected]>

catchup/universalFetcher_test.go

@@ -204,7 +204,7 @@ func TestRequestBlockBytesErrors(t *testing.T) {
 	cancel()
 	_, _, _, err = fetcher.fetchBlock(ctx, next, up)
 	var wrfe errWsFetcherRequestFailed
-	require.True(t, errors.As(err, &wrfe), "unexpected err: %w", wrfe)
+	require.ErrorAs(t, err, &wrfe)
 	require.Equal(t, "context canceled", err.(errWsFetcherRequestFailed).cause)
 
 	ctx = context.Background()
@@ -213,14 +213,14 @@ func TestRequestBlockBytesErrors(t *testing.T) {
 	up = makeTestUnicastPeerWithResponseOverride(net, t, &responseOverride)
 
 	_, _, _, err = fetcher.fetchBlock(ctx, next, up)
-	require.True(t, errors.As(err, &wrfe))
+	require.ErrorAs(t, err, &wrfe)
 	require.Equal(t, "Cert data not found", err.(errWsFetcherRequestFailed).cause)
 
 	responseOverride = network.Response{Topics: network.Topics{network.MakeTopic(rpcs.CertDataKey, make([]byte, 0))}}
 	up = makeTestUnicastPeerWithResponseOverride(net, t, &responseOverride)
 
 	_, _, _, err = fetcher.fetchBlock(ctx, next, up)
-	require.True(t, errors.As(err, &wrfe))
+	require.ErrorAs(t, err, &wrfe)
 	require.Equal(t, "Block data not found", err.(errWsFetcherRequestFailed).cause)
 }
 
@@ -240,7 +240,6 @@ func (thh *TestHTTPHandler) ServeHTTP(response http.ResponseWriter, request *htt
 		bytes = make([]byte, fetcherMaxBlockBytes+1)
 	}
 	response.Write(bytes)
-	return
 }
 
 // TestGetBlockBytesHTTPErrors tests the errors reported from getblockBytes for http peer
@@ -264,25 +263,25 @@ func TestGetBlockBytesHTTPErrors(t *testing.T) {
 	ls.status = http.StatusBadRequest
 	_, _, _, err := fetcher.fetchBlock(context.Background(), 1, net.GetPeers()[0])
 	var hre errHTTPResponse
-	require.True(t, errors.As(err, &hre))
+	require.ErrorAs(t, err, &hre)
 	require.Equal(t, "Response body '\x00'", err.(errHTTPResponse).cause)
 
 	ls.exceedLimit = true
 	_, _, _, err = fetcher.fetchBlock(context.Background(), 1, net.GetPeers()[0])
-	require.True(t, errors.As(err, &hre))
+	require.ErrorAs(t, err, &hre)
 	require.Equal(t, "read limit exceeded", err.(errHTTPResponse).cause)
 
 	ls.status = http.StatusOK
 	ls.content = append(ls.content, "undefined")
 	_, _, _, err = fetcher.fetchBlock(context.Background(), 1, net.GetPeers()[0])
 	var cte errHTTPResponseContentType
-	require.True(t, errors.As(err, &cte))
+	require.ErrorAs(t, err, &cte)
 	require.Equal(t, "undefined", err.(errHTTPResponseContentType).contentType)
 
 	ls.status = http.StatusOK
 	ls.content = append(ls.content, "undefined2")
 	_, _, _, err = fetcher.fetchBlock(context.Background(), 1, net.GetPeers()[0])
-	require.True(t, errors.As(err, &cte))
+	require.ErrorAs(t, err, &cte)
 	require.Equal(t, 2, err.(errHTTPResponseContentType).contentTypeCount)
 }
 

network/p2pNetwork.go

@@ -808,6 +808,12 @@ func (n *P2PNetwork) wsStreamHandler(ctx context.Context, p2pPeer peer.ID, strea
 
 	wsp.init(n.config, outgoingMessagesBufferSize)
 	n.wsPeersLock.Lock()
+	if wsp.didSignalClose.Load() == 1 {
+		networkPeerAlreadyClosed.Inc(nil)
+		n.log.Debugf("peer closing %s", addr)
+		n.wsPeersLock.Unlock()
+		return
+	}
 	n.wsPeers[p2pPeer] = wsp
 	n.wsPeersToIDs[wsp] = p2pPeer
 	n.wsPeersLock.Unlock()