We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Dark Reaper rse_id crash
Motivation
----------
The dark reaper raises an error in 1.20.3


Modification
------------
Fix the query in core/quarantined_replica.py -> list_rses()



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lib/rucio/core/quarantined_replica.py`
Content:
```
1 #!/usr/bin/env python
2 # Copyright 2016-2018 CERN for the benefit of the ATLAS collaboration.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #    http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 # Authors:
17 # - Vincent Garonne <vgaronne@gmail.com>, 2016-2017
18 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018
19 # - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018
20 # - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019
21 #
22 # PY3K COMPATIBLE
23 
24 import datetime
25 
26 from sqlalchemy import and_, or_, exists, not_
27 from sqlalchemy.sql.expression import bindparam, text, select, false
28 
29 from rucio.common.utils import chunks
30 from rucio.db.sqla import models
31 from rucio.db.sqla.session import read_session, transactional_session
32 
33 
34 @transactional_session
35 def add_quarantined_replicas(rse_id, replicas, session=None):
36     """
37     Bulk add quarantined file replicas.
38 
39     :param rse_id:      The rse id.
40     :param replicas: A list of dicts with the replica information.
41     :param session:  The database session in use.
42     """
43 
44     for chunk in chunks(replicas, 100):
45         # Exlude files that have a registered replica.  This is a
46         # safeguard against potential issues in the Auditor.
47         file_clause = []
48         for replica in chunk:
49             file_clause.append(and_(models.RSEFileAssociation.scope == replica.get('scope', None),
50                                     models.RSEFileAssociation.name == replica.get('name', None),
51                                     models.RSEFileAssociation.rse_id == rse_id))
52         file_query = session.query(models.RSEFileAssociation.scope,
53                                    models.RSEFileAssociation.name,
54                                    models.RSEFileAssociation.rse_id).\
55             with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\
56             filter(or_(*file_clause))
57         existing_replicas = [(scope, name, rseid) for scope, name, rseid in file_query]
58         chunk = [replica for replica in chunk if (replica.get('scope', None), replica.get('name', None), rse_id) not in existing_replicas]
59 
60         # Exclude files that have already been added to the quarantined
61         # replica table.
62         quarantine_clause = []
63         for replica in chunk:
64             quarantine_clause.append(and_(models.QuarantinedReplica.path == replica['path'],
65                                           models.QuarantinedReplica.rse_id == rse_id))
66         quarantine_query = session.query(models.QuarantinedReplica.path,
67                                          models.QuarantinedReplica.rse_id).\
68             filter(or_(*quarantine_clause))
69         quarantine_replicas = [(path, rseid) for path, rseid in quarantine_query]
70         chunk = [replica for replica in chunk if (replica['path'], rse_id) not in quarantine_replicas]
71 
72         session.bulk_insert_mappings(
73             models.QuarantinedReplica,
74             [{'rse_id': rse_id, 'path': file['path'],
75               'scope': file.get('scope'), 'name': file.get('name'),
76               'bytes': file.get('bytes')} for file in chunk])
77 
78 
79 @transactional_session
80 def delete_quarantined_replicas(rse_id, replicas, session=None):
81     """
82     Delete file replicas.
83 
84     :param rse_id: the rse id.
85     :param files: the list of files to delete.
86     :param ignore_availability: Ignore the RSE blacklisting.
87     :param session: The database session in use.
88     """
89 
90     conditions = []
91     for replica in replicas:
92         conditions.append(models.QuarantinedReplica.path == replica['path'])
93 
94     if conditions:
95         session.query(models.QuarantinedReplica).\
96             filter(models.QuarantinedReplica.rse_id == rse_id).\
97             filter(or_(*conditions)).\
98             delete(synchronize_session=False)
99 
100     session.\
101         bulk_insert_mappings(models.QuarantinedReplica.__history_mapper__.class_,
102                              [{'rse_id': rse_id, 'path': replica['path'],
103                                'bytes': replica.get('bytes'),
104                                'created_at': replica.get('created_at'),
105                                'deleted_at': datetime.datetime.utcnow()}
106                               for replica in replicas])
107 
108 
109 @read_session
110 def list_quarantined_replicas(rse_id, limit, worker_number=None, total_workers=None, session=None):
111     """
112     List RSE Quarantined File replicas.
113 
114     :param rse_id: the rse id.
115     :param limit: The maximum number of replicas returned.
116     :param worker_number:      id of the executing worker.
117     :param total_workers:      Number of total workers.
118     :param session: The database session in use.
119 
120     :returns: a list of dictionary replica.
121     """
122 
123     query = session.query(models.QuarantinedReplica.path,
124                           models.QuarantinedReplica.bytes,
125                           models.QuarantinedReplica.scope,
126                           models.QuarantinedReplica.name,
127                           models.QuarantinedReplica.created_at).\
128         filter(models.QuarantinedReplica.rse_id == rse_id)
129 
130     # do no delete valid replicas
131     stmt = exists(select([1]).prefix_with("/*+ index(REPLICAS REPLICAS_PK) */", dialect='oracle')).\
132         where(and_(models.RSEFileAssociation.scope == models.QuarantinedReplica.scope,
133                    models.RSEFileAssociation.name == models.QuarantinedReplica.name,
134                    models.RSEFileAssociation.rse_id == models.QuarantinedReplica.rse_id))
135     query = query.filter(not_(stmt))
136 
137     if worker_number and total_workers and total_workers - 1 > 0:
138         if session.bind.dialect.name == 'oracle':
139             bindparams = [bindparam('worker_number', worker_number - 1), bindparam('total_workers', total_workers - 1)]
140             query = query.filter(text('ORA_HASH(path, :total_workers) = :worker_number', bindparams=bindparams))
141         elif session.bind.dialect.name == 'mysql':
142             query = query.filter('mod(md5(path), %s) = %s' % (total_workers - 1, worker_number - 1))
143         elif session.bind.dialect.name == 'postgresql':
144             query = query.filter('mod(abs((\'x\'||md5(path))::bit(32)::int), %s) = %s' % (total_workers - 1, worker_number - 1))
145 
146     return [{'path': path,
147              'rse_id': rse_id,
148              'created_at': created_at,
149              'scope': scope,
150              'name': name,
151              'bytes': bytes}
152             for path, bytes, scope, name, created_at in query.limit(limit)]
153 
154 
155 @read_session
156 def list_rses(session=None):
157     """
158     List RSEs in the Quarantined Queues.
159 
160     :param session: The database session in use.
161 
162     :returns: a list of RSEs.
163     """
164     query = session.query(models.RSE.rse_id).distinct(models.RSE.rse_id).\
165         filter(models.QuarantinedReplica.rse_id == models.RSE.id).\
166         filter(models.RSE.deleted == false())
167     return [rse for (rse,) in query]
168
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lib/rucio/core/quarantined_replica.py b/lib/rucio/core/quarantined_replica.py
--- a/lib/rucio/core/quarantined_replica.py
+++ b/lib/rucio/core/quarantined_replica.py
@@ -161,7 +161,7 @@
 
     :returns: a list of RSEs.
     """
-    query = session.query(models.RSE.rse_id).distinct(models.RSE.rse_id).\
+    query = session.query(models.RSE.id).distinct(models.RSE.id).\
         filter(models.QuarantinedReplica.rse_id == models.RSE.id).\
         filter(models.RSE.deleted == false())
     return [rse for (rse,) in query]

{"golden_diff": "diff --git a/lib/rucio/core/quarantined_replica.py b/lib/rucio/core/quarantined_replica.py\n--- a/lib/rucio/core/quarantined_replica.py\n+++ b/lib/rucio/core/quarantined_replica.py\n@@ -161,7 +161,7 @@\n \n     :returns: a list of RSEs.\n     \"\"\"\n-    query = session.query(models.RSE.rse_id).distinct(models.RSE.rse_id).\\\n+    query = session.query(models.RSE.id).distinct(models.RSE.id).\\\n         filter(models.QuarantinedReplica.rse_id == models.RSE.id).\\\n         filter(models.RSE.deleted == false())\n     return [rse for (rse,) in query]\n", "issue": "Dark Reaper rse_id crash\nMotivation\r\n----------\r\nThe dark reaper raises an error in 1.20.3\r\n\r\n\r\nModification\r\n------------\r\nFix the query in core/quarantined_replica.py -> list_rses()\r\n\r\n\n", "before_files": [{"content": "#!/usr/bin/env python\n# Copyright 2016-2018 CERN for the benefit of the ATLAS collaboration.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n# Authors:\n# - Vincent Garonne <vgaronne@gmail.com>, 2016-2017\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018\n# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018\n# - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019\n#\n# PY3K COMPATIBLE\n\nimport datetime\n\nfrom sqlalchemy import and_, or_, exists, not_\nfrom sqlalchemy.sql.expression import bindparam, text, select, false\n\nfrom rucio.common.utils import chunks\nfrom rucio.db.sqla import models\nfrom rucio.db.sqla.session import read_session, transactional_session\n\n\n@transactional_session\ndef add_quarantined_replicas(rse_id, replicas, session=None):\n    \"\"\"\n    Bulk add quarantined file replicas.\n\n    :param rse_id:      The rse id.\n    :param replicas: A list of dicts with the replica information.\n    :param session:  The database session in use.\n    \"\"\"\n\n    for chunk in chunks(replicas, 100):\n        # Exlude files that have a registered replica.  This is a\n        # safeguard against potential issues in the Auditor.\n        file_clause = []\n        for replica in chunk:\n            file_clause.append(and_(models.RSEFileAssociation.scope == replica.get('scope', None),\n                                    models.RSEFileAssociation.name == replica.get('name', None),\n                                    models.RSEFileAssociation.rse_id == rse_id))\n        file_query = session.query(models.RSEFileAssociation.scope,\n                                   models.RSEFileAssociation.name,\n                                   models.RSEFileAssociation.rse_id).\\\n            with_hint(models.RSEFileAssociation, \"index(REPLICAS REPLICAS_PK)\", 'oracle').\\\n            filter(or_(*file_clause))\n        existing_replicas = [(scope, name, rseid) for scope, name, rseid in file_query]\n        chunk = [replica for replica in chunk if (replica.get('scope', None), replica.get('name', None), rse_id) not in existing_replicas]\n\n        # Exclude files that have already been added to the quarantined\n        # replica table.\n        quarantine_clause = []\n        for replica in chunk:\n            quarantine_clause.append(and_(models.QuarantinedReplica.path == replica['path'],\n                                          models.QuarantinedReplica.rse_id == rse_id))\n        quarantine_query = session.query(models.QuarantinedReplica.path,\n                                         models.QuarantinedReplica.rse_id).\\\n            filter(or_(*quarantine_clause))\n        quarantine_replicas = [(path, rseid) for path, rseid in quarantine_query]\n        chunk = [replica for replica in chunk if (replica['path'], rse_id) not in quarantine_replicas]\n\n        session.bulk_insert_mappings(\n            models.QuarantinedReplica,\n            [{'rse_id': rse_id, 'path': file['path'],\n              'scope': file.get('scope'), 'name': file.get('name'),\n              'bytes': file.get('bytes')} for file in chunk])\n\n\n@transactional_session\ndef delete_quarantined_replicas(rse_id, replicas, session=None):\n    \"\"\"\n    Delete file replicas.\n\n    :param rse_id: the rse id.\n    :param files: the list of files to delete.\n    :param ignore_availability: Ignore the RSE blacklisting.\n    :param session: The database session in use.\n    \"\"\"\n\n    conditions = []\n    for replica in replicas:\n        conditions.append(models.QuarantinedReplica.path == replica['path'])\n\n    if conditions:\n        session.query(models.QuarantinedReplica).\\\n            filter(models.QuarantinedReplica.rse_id == rse_id).\\\n            filter(or_(*conditions)).\\\n            delete(synchronize_session=False)\n\n    session.\\\n        bulk_insert_mappings(models.QuarantinedReplica.__history_mapper__.class_,\n                             [{'rse_id': rse_id, 'path': replica['path'],\n                               'bytes': replica.get('bytes'),\n                               'created_at': replica.get('created_at'),\n                               'deleted_at': datetime.datetime.utcnow()}\n                              for replica in replicas])\n\n\n@read_session\ndef list_quarantined_replicas(rse_id, limit, worker_number=None, total_workers=None, session=None):\n    \"\"\"\n    List RSE Quarantined File replicas.\n\n    :param rse_id: the rse id.\n    :param limit: The maximum number of replicas returned.\n    :param worker_number:      id of the executing worker.\n    :param total_workers:      Number of total workers.\n    :param session: The database session in use.\n\n    :returns: a list of dictionary replica.\n    \"\"\"\n\n    query = session.query(models.QuarantinedReplica.path,\n                          models.QuarantinedReplica.bytes,\n                          models.QuarantinedReplica.scope,\n                          models.QuarantinedReplica.name,\n                          models.QuarantinedReplica.created_at).\\\n        filter(models.QuarantinedReplica.rse_id == rse_id)\n\n    # do no delete valid replicas\n    stmt = exists(select([1]).prefix_with(\"/*+ index(REPLICAS REPLICAS_PK) */\", dialect='oracle')).\\\n        where(and_(models.RSEFileAssociation.scope == models.QuarantinedReplica.scope,\n                   models.RSEFileAssociation.name == models.QuarantinedReplica.name,\n                   models.RSEFileAssociation.rse_id == models.QuarantinedReplica.rse_id))\n    query = query.filter(not_(stmt))\n\n    if worker_number and total_workers and total_workers - 1 > 0:\n        if session.bind.dialect.name == 'oracle':\n            bindparams = [bindparam('worker_number', worker_number - 1), bindparam('total_workers', total_workers - 1)]\n            query = query.filter(text('ORA_HASH(path, :total_workers) = :worker_number', bindparams=bindparams))\n        elif session.bind.dialect.name == 'mysql':\n            query = query.filter('mod(md5(path), %s) = %s' % (total_workers - 1, worker_number - 1))\n        elif session.bind.dialect.name == 'postgresql':\n            query = query.filter('mod(abs((\\'x\\'||md5(path))::bit(32)::int), %s) = %s' % (total_workers - 1, worker_number - 1))\n\n    return [{'path': path,\n             'rse_id': rse_id,\n             'created_at': created_at,\n             'scope': scope,\n             'name': name,\n             'bytes': bytes}\n            for path, bytes, scope, name, created_at in query.limit(limit)]\n\n\n@read_session\ndef list_rses(session=None):\n    \"\"\"\n    List RSEs in the Quarantined Queues.\n\n    :param session: The database session in use.\n\n    :returns: a list of RSEs.\n    \"\"\"\n    query = session.query(models.RSE.rse_id).distinct(models.RSE.rse_id).\\\n        filter(models.QuarantinedReplica.rse_id == models.RSE.id).\\\n        filter(models.RSE.deleted == false())\n    return [rse for (rse,) in query]\n", "path": "lib/rucio/core/quarantined_replica.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# Copyright 2016-2018 CERN for the benefit of the ATLAS collaboration.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n# Authors:\n# - Vincent Garonne <vgaronne@gmail.com>, 2016-2017\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018\n# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018\n# - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019\n#\n# PY3K COMPATIBLE\n\nimport datetime\n\nfrom sqlalchemy import and_, or_, exists, not_\nfrom sqlalchemy.sql.expression import bindparam, text, select, false\n\nfrom rucio.common.utils import chunks\nfrom rucio.db.sqla import models\nfrom rucio.db.sqla.session import read_session, transactional_session\n\n\n@transactional_session\ndef add_quarantined_replicas(rse_id, replicas, session=None):\n    \"\"\"\n    Bulk add quarantined file replicas.\n\n    :param rse_id:      The rse id.\n    :param replicas: A list of dicts with the replica information.\n    :param session:  The database session in use.\n    \"\"\"\n\n    for chunk in chunks(replicas, 100):\n        # Exlude files that have a registered replica.  This is a\n        # safeguard against potential issues in the Auditor.\n        file_clause = []\n        for replica in chunk:\n            file_clause.append(and_(models.RSEFileAssociation.scope == replica.get('scope', None),\n                                    models.RSEFileAssociation.name == replica.get('name', None),\n                                    models.RSEFileAssociation.rse_id == rse_id))\n        file_query = session.query(models.RSEFileAssociation.scope,\n                                   models.RSEFileAssociation.name,\n                                   models.RSEFileAssociation.rse_id).\\\n            with_hint(models.RSEFileAssociation, \"index(REPLICAS REPLICAS_PK)\", 'oracle').\\\n            filter(or_(*file_clause))\n        existing_replicas = [(scope, name, rseid) for scope, name, rseid in file_query]\n        chunk = [replica for replica in chunk if (replica.get('scope', None), replica.get('name', None), rse_id) not in existing_replicas]\n\n        # Exclude files that have already been added to the quarantined\n        # replica table.\n        quarantine_clause = []\n        for replica in chunk:\n            quarantine_clause.append(and_(models.QuarantinedReplica.path == replica['path'],\n                                          models.QuarantinedReplica.rse_id == rse_id))\n        quarantine_query = session.query(models.QuarantinedReplica.path,\n                                         models.QuarantinedReplica.rse_id).\\\n            filter(or_(*quarantine_clause))\n        quarantine_replicas = [(path, rseid) for path, rseid in quarantine_query]\n        chunk = [replica for replica in chunk if (replica['path'], rse_id) not in quarantine_replicas]\n\n        session.bulk_insert_mappings(\n            models.QuarantinedReplica,\n            [{'rse_id': rse_id, 'path': file['path'],\n              'scope': file.get('scope'), 'name': file.get('name'),\n              'bytes': file.get('bytes')} for file in chunk])\n\n\n@transactional_session\ndef delete_quarantined_replicas(rse_id, replicas, session=None):\n    \"\"\"\n    Delete file replicas.\n\n    :param rse_id: the rse id.\n    :param files: the list of files to delete.\n    :param ignore_availability: Ignore the RSE blacklisting.\n    :param session: The database session in use.\n    \"\"\"\n\n    conditions = []\n    for replica in replicas:\n        conditions.append(models.QuarantinedReplica.path == replica['path'])\n\n    if conditions:\n        session.query(models.QuarantinedReplica).\\\n            filter(models.QuarantinedReplica.rse_id == rse_id).\\\n            filter(or_(*conditions)).\\\n            delete(synchronize_session=False)\n\n    session.\\\n        bulk_insert_mappings(models.QuarantinedReplica.__history_mapper__.class_,\n                             [{'rse_id': rse_id, 'path': replica['path'],\n                               'bytes': replica.get('bytes'),\n                               'created_at': replica.get('created_at'),\n                               'deleted_at': datetime.datetime.utcnow()}\n                              for replica in replicas])\n\n\n@read_session\ndef list_quarantined_replicas(rse_id, limit, worker_number=None, total_workers=None, session=None):\n    \"\"\"\n    List RSE Quarantined File replicas.\n\n    :param rse_id: the rse id.\n    :param limit: The maximum number of replicas returned.\n    :param worker_number:      id of the executing worker.\n    :param total_workers:      Number of total workers.\n    :param session: The database session in use.\n\n    :returns: a list of dictionary replica.\n    \"\"\"\n\n    query = session.query(models.QuarantinedReplica.path,\n                          models.QuarantinedReplica.bytes,\n                          models.QuarantinedReplica.scope,\n                          models.QuarantinedReplica.name,\n                          models.QuarantinedReplica.created_at).\\\n        filter(models.QuarantinedReplica.rse_id == rse_id)\n\n    # do no delete valid replicas\n    stmt = exists(select([1]).prefix_with(\"/*+ index(REPLICAS REPLICAS_PK) */\", dialect='oracle')).\\\n        where(and_(models.RSEFileAssociation.scope == models.QuarantinedReplica.scope,\n                   models.RSEFileAssociation.name == models.QuarantinedReplica.name,\n                   models.RSEFileAssociation.rse_id == models.QuarantinedReplica.rse_id))\n    query = query.filter(not_(stmt))\n\n    if worker_number and total_workers and total_workers - 1 > 0:\n        if session.bind.dialect.name == 'oracle':\n            bindparams = [bindparam('worker_number', worker_number - 1), bindparam('total_workers', total_workers - 1)]\n            query = query.filter(text('ORA_HASH(path, :total_workers) = :worker_number', bindparams=bindparams))\n        elif session.bind.dialect.name == 'mysql':\n            query = query.filter('mod(md5(path), %s) = %s' % (total_workers - 1, worker_number - 1))\n        elif session.bind.dialect.name == 'postgresql':\n            query = query.filter('mod(abs((\\'x\\'||md5(path))::bit(32)::int), %s) = %s' % (total_workers - 1, worker_number - 1))\n\n    return [{'path': path,\n             'rse_id': rse_id,\n             'created_at': created_at,\n             'scope': scope,\n             'name': name,\n             'bytes': bytes}\n            for path, bytes, scope, name, created_at in query.limit(limit)]\n\n\n@read_session\ndef list_rses(session=None):\n    \"\"\"\n    List RSEs in the Quarantined Queues.\n\n    :param session: The database session in use.\n\n    :returns: a list of RSEs.\n    \"\"\"\n    query = session.query(models.RSE.id).distinct(models.RSE.id).\\\n        filter(models.QuarantinedReplica.rse_id == models.RSE.id).\\\n        filter(models.RSE.deleted == false())\n    return [rse for (rse,) in query]\n", "path": "lib/rucio/core/quarantined_replica.py"}]}