get_pfn_to_rse() doesn’t properly handle protocols with different ports
Motivation
----------

In ATLAS, we have RSEs with protocols that have the same scheme, same domain, but different port and activities. They appear to be creating problems during bad file declaration. In this case, BNL has WebDAV running on `dcgftp.usatlas.bnl.gov` on ports `443` and `2881`, the former for all activities except deletion, the latter only for deletion.

`rucio list-file-replicas` returns URLs with port `443`. These URLs cannot be used for lost file declaration. Removing the port number succeeds.

```python
>>> from rucio.core.replica import get_pfn_to_rse
>>> get_pfn_to_rse(['davs://dcgftp.usatlas.bnl.gov:443/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/mc16_13TeV/22/6d/log.27574092._000437.job.log.tgz.1'])
('davs', {}, {'unknown': ['davs://dcgftp.usatlas.bnl.gov:443/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/mc16_13TeV/22/6d/log.27574092._000437.job.log.tgz.1']})
>>> get_pfn_to_rse(['davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/mc16_13TeV/22/6d/log.27574092._000437.job.log.tgz.1'])
('davs', {'b64daf80eed64d5aabb02a6c457edc44': ['davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/mc16_13TeV/22/6d/log.27574092._000437.job.log.tgz.1']}, {})
```

If one checks the structure built internally:

https://github.com/rucio/rucio/blob/80f8c5a96e06abb80e3e6c854b89298f16be6270/lib/rucio/core/replica.py#L527-L528

```python
>>> pprint.pprint(protocols)
{'0d0628a4b40e4ddb8b088e848177827d': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/lake_localgroupdisk/rucio',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/lake_localgroupdisk/rucio'),
 '780d557dc3164bceb1cb26d07607571f': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/lake/rucio',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/lake/rucio'),
 'ac992742a7b542f9953bbb1a7dc43b91': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/atlasscratchdisk/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/atlasscratchdisk/rucio/'),
 'b64daf80eed64d5aabb02a6c457edc44': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/BNLT0D1/rucio/'),
 'd456ce95a59d4e46b99f26a5459ac14a': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/LOCALGROUPDISK/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/LOCALGROUPDISK/rucio/'),
 'd50cf8c3ec6b4c3dbdb894e09910d2fe': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/atlasvpdisk/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/atlasvpdisk/rucio/'),
 'd7c73c46d06e480e851dc655a473719c': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/atlasgroupdisk/phys-hi/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/atlasgroupdisk/phys-hi/rucio/'),
 'e753b2f096814143b7f908d8c343b5b2': ('davs://dcgftp.usatlas.bnl.gov/pnfs/usatlas.bnl.gov/atlasgroupdisk/det-slhc/rucio/',
                                      'davs://dcgftp.usatlas.bnl.gov:2881/pnfs/usatlas.bnl.gov/atlasgroupdisk/det-slhc/rucio/')}
```

The order of the rows, as returned by the database, matters and protocols with `443` were overwritten by the ones with `2881`.

Modification
------------

diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 0b6b89a6c..590a2ee7b 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -459,23 +459,30 @@ def get_pfn_to_rse(pfns, vo='def', session=None):
     protocols = {}
 
     for rse_id, protocol, hostname, port, prefix in query.yield_per(10000):
-        protocols[rse_id] = ('%s://%s%s' % (protocol, hostname, prefix), '%s://%s:%s%s' % (protocol, hostname, port, prefix))
+        if rse_id not in protocols:
+            protocols[rse_id] = []
+        protocols[rse_id].append('%s://%s:%s%s' % (protocol, hostname, port, prefix))
+        if '%s://%s%s' % (protocol, hostname, prefix) not in protocols[rse_id]:
+            protocols[rse_id].append('%s://%s%s' % (protocol, hostname, prefix))
     hint = None
     for surl in surls:
-        if hint and (surl.find(protocols[hint][0]) > -1 or surl.find(protocols[hint][1]) > -1):
-            dict_rse[hint].append(surl)
+        if hint:
+            for pattern in protocols[hint]:
+                if surl.find(pattern) > -1:
+                    dict_rse[hint].append(surl)
         else:
             mult_rse_match = 0
             for rse_id in protocols:
-                if (surl.find(protocols[rse_id][0]) > -1 or surl.find(protocols[rse_id][1]) > -1) and get_rse_vo(rse_id=rse_id, session=session) == vo:
-                    mult_rse_match += 1
-                    if mult_rse_match > 1:
-                        print('ERROR, multiple matches : %s at %s' % (surl, rse_id))
-                        raise exception.RucioException('ERROR, multiple matches : %s at %s' % (surl, get_rse_name(rse_id=rse_id, session=session)))
-                    hint = rse_id
-                    if hint not in dict_rse:
-                        dict_rse[hint] = []
-                    dict_rse[hint].append(surl)
+                for pattern in protocols[rse_id]:
+                    if surl.find(pattern) > -1 and get_rse_vo(rse_id=rse_id, session=session) == vo:
+                        mult_rse_match += 1
+                        if mult_rse_match > 1:
+                            print('ERROR, multiple matches : %s at %s' % (surl, rse_id))
+                            raise exception.RucioException('ERROR, multiple matches : %s at %s' % (surl, get_rse_name(rse_id=rse_id, session=session)))
+                        hint = rse_id
+                        if hint not in dict_rse:
+                            dict_rse[hint] = []
+                        dict_rse[hint].append(surl)
             if mult_rse_match == 0:
                 if 'unknown' not in unknown_replicas:
                     unknown_replicas['unknown'] = []
diff --git a/lib/rucio/tests/test_bad_replica.py b/lib/rucio/tests/test_bad_replica.py
index 2c3f8d093..9926c67a6 100644
--- a/lib/rucio/tests/test_bad_replica.py
+++ b/lib/rucio/tests/test_bad_replica.py
@@ -25,7 +25,8 @@ from rucio.common.utils import generate_uuid, clean_surls
 from rucio.core.did import delete_dids
 from rucio.core.replica import (add_replicas, get_replicas_state, list_replicas,
                                 declare_bad_file_replicas, list_bad_replicas, get_bad_pfns,
-                                get_bad_replicas_backlog, list_bad_replicas_status)
+                                get_bad_replicas_backlog, list_bad_replicas_status, get_pfn_to_rse)
+from rucio.client.rseclient import RSEClient
 from rucio.daemons.badreplicas.minos import run as minos_run
 from rucio.daemons.badreplicas.minos_temporary_expiration import run as minos_temp_run
 from rucio.daemons.badreplicas.necromancer import run as necromancer_run
@@ -34,6 +35,11 @@ from rucio.db.sqla.constants import DIDType, ReplicaState, BadPFNStatus, BadFile
 from rucio.tests.common import headers, auth
 
 
+@pytest.fixture
+def rse_client():
+    return RSEClient()
+
+
 @pytest.mark.noparallel(reason='calls list_bad_replicas() which acts on all bad replicas without any filtering')
 def test_add_list_bad_replicas(rse_factory, mock_scope, root_account):
     """ REPLICA (CORE): Add bad replicas and list them"""
@@ -457,3 +463,47 @@ def test_add_and_delete_bad_replicas(rse_factory, mock_scope, root_account, did_
     list_deleted_rep = [{'scope': rep['scope'].external, 'name': rep['name']} for rep in list_bad_replicas_status(state=BadFilesStatus.DELETED, rse_id=rse1_id, vo=vo)]
     for rep in client_files:
         assert rep in list_deleted_rep
+
+
+def test_get_pfn_to_rse(rse_factory, rse_client, vo):
+    """ REPLICA (CORE): Test that get_pfn_to_rse is able to handle same scheme with different port"""
+
+    # Adding replicas to deterministic RSE
+    rse1, rse1_id = rse_factory.make_rse(scheme='file', protocol_impl='rucio.rse.protocols.posix.Default', vo=vo)
+    protocols = [{'scheme': 'MOCK',
+                  'hostname': 'localhost',
+                  'port': 17,
+                  'prefix': '/the/one/with/all/the/files',
+                  'impl': 'rucio.rse.protocols.SomeProtocol.SomeImplementation',
+                  'domains': {
+                      'lan': {'read': 4,
+                              'write': 1,
+                              'delete': 0}
+                  },
+                  'extended_attributes': 'TheOneWithAllTheRest'},
+                 {'scheme': 'MOCK',
+                  'hostname': 'localhost',
+                  'port': 18,
+                  'prefix': '/the/one/with/all/the/files',
+                  'impl': 'rucio.rse.protocols.SomeProtocol.SomeImplementation',
+                  'domains': {
+                      'lan': {'read': 1,
+                              'write': 1,
+                              'delete': 0}},
+                  'extended_attributes': 'TheOneWithAllTheRest'}, ]
+    for prot in protocols:
+        rse_client.add_protocol(rse1, prot)
+
+    pfn = 'MOCK://localhost:17/the/one/with/all/the/files/file1'
+    pfn_no_port = 'MOCK://localhost/the/one/with/all/the/files/file1'
+    pfn_other_port = 'MOCK://localhost:18/the/one/with/all/the/files/file1'
+    pfn_non_existing_port = 'MOCK://localhost:19/the/one/with/all/the/files/file1'
+
+    res = get_pfn_to_rse([pfn], vo=vo)
+    assert res == ('MOCK', {rse1_id: [pfn]}, {})
+    res = get_pfn_to_rse([pfn_no_port], vo=vo)
+    assert res == ('MOCK', {rse1_id: [pfn_no_port]}, {})
+    res = get_pfn_to_rse([pfn_other_port], vo=vo)
+    assert res == ('MOCK', {rse1_id: [pfn_other_port]}, {})
+    res = get_pfn_to_rse([pfn_non_existing_port], vo=vo)
+    assert res == ('MOCK', {}, {'unknown': [pfn_non_existing_port]})
['lib/rucio/tests/test_bad_replica.py', 'lib/rucio/core/replica.py']