LOST transfers incorrectly handled when FTS returns 404 status code
Motivation
----------
For non-existing transfers, FTS can either return a 207, or a 404, status code. This depends on whether all queried transfers are lost, or only some of them. Rucio incorrectly handles the 404 status code. It is registered as "failure" instead of lost transfers. This is not problematic, because the end result is the same: transfers are re-submitted, but it's worth fixing the workflow. 


diff --git a/lib/rucio/tests/test_conveyor.py b/lib/rucio/tests/test_conveyor.py
index bab13db57..92dc17e59 100644
--- a/lib/rucio/tests/test_conveyor.py
+++ b/lib/rucio/tests/test_conveyor.py
@@ -20,7 +20,7 @@
 
 import threading
 import time
-from datetime import datetime
+from datetime import datetime, timedelta
 from unittest.mock import patch
 from urllib.parse import urlencode, urlparse, parse_qsl, urlunparse
 
@@ -618,6 +618,46 @@ def test_stager(rse_factory, did_factory, root_account, replica_client):
     assert replica['state'] == ReplicaState.AVAILABLE
 
 
+@skip_rse_tests_with_accounts
+@pytest.mark.noparallel(reason="runs submitter; poller and finisher")
+def test_lost_transfers(rse_factory, did_factory, root_account):
+    """
+    Correctly handle FTS "404 not found" errors.
+    """
+    src_rse, src_rse_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    dst_rse, dst_rse_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    all_rses = [src_rse_id, dst_rse_id]
+
+    distance_core.add_distance(src_rse_id, dst_rse_id, ranking=10)
+    for rse_id in all_rses:
+        rse_core.add_rse_attribute(rse_id, 'fts', TEST_FTS_HOST)
+
+    did = did_factory.upload_test_file(src_rse)
+
+    rule_core.add_rule(dids=[did], account=root_account, copies=1, rse_expression=dst_rse, grouping='ALL', weight=None, lifetime=None, locked=False, subscription_id=None)
+
+    @transactional_session
+    def __update_request(request_id, session=None, **kwargs):
+        session.query(models.Request).filter_by(id=request_id).update(kwargs, synchronize_session=False)
+
+    # Fake that the transfer is submitted and lost
+    submitter(once=True, rses=[{'id': rse_id} for rse_id in all_rses], group_bulk=2, partition_wait_time=None, transfertype='single', filter_transfertool=None)
+    request = request_core.get_request_by_did(rse_id=dst_rse_id, **did)
+    __update_request(request['id'], external_id='some-fake-random-id')
+
+    # The request must be marked lost
+    request = __wait_for_request_state(dst_rse_id=dst_rse_id, state=RequestState.LOST, **did)
+    assert request['state'] == RequestState.LOST
+
+    # Set update time far in the past to bypass protections (not resubmitting too fast).
+    # Run finisher and submitter, the request must be resubmitted and transferred correctly
+    __update_request(request['id'], updated_at=datetime.utcnow() - timedelta(days=1))
+    finisher(once=True, partition_wait_time=None)
+    submitter(once=True, rses=[{'id': rse_id} for rse_id in all_rses], group_bulk=2, partition_wait_time=None, transfertype='single', filter_transfertool=None)
+    replica = __wait_for_replica_transfer(dst_rse_id=dst_rse_id, **did)
+    assert replica['state'] == ReplicaState.AVAILABLE
+
+
 @skip_rse_tests_with_accounts
 @pytest.mark.noparallel(reason="runs submitter and poller;")
 @pytest.mark.parametrize("core_config_mock", [{"table_content": [
diff --git a/lib/rucio/transfertool/fts3.py b/lib/rucio/transfertool/fts3.py
index 379d909e5..0cfc4748e 100644
--- a/lib/rucio/transfertool/fts3.py
+++ b/lib/rucio/transfertool/fts3.py
@@ -398,7 +398,7 @@ class FTS3Transfertool(Transfertool):
             record_counter('transfertool.fts3.{host}.bulk_query.failure', labels={'host': self.__extract_host(self.external_host)})
             for transfer_id in transfer_ids:
                 responses[transfer_id] = Exception('Transfer information returns None: %s' % jobs)
-        elif jobs.status_code == 200 or jobs.status_code == 207:
+        elif jobs.status_code in (200, 207, 404):
             try:
                 BULK_QUERY_COUNTER.labels(state='success', host=self.__extract_host(self.external_host)).inc()
                 jobs_response = jobs.json()
['lib/rucio/transfertool/fts3.py', 'lib/rucio/tests/test_conveyor.py']