Repository: rucio/rucio
Issue: source URL not updated correctly for multi-source transfers

Motivation
----------

For multi source transfers the request inside Rucio is pre-filled with one of the possible sources. FTS then might chose a different source from the list and will report back which one was used. The poller then updates the initial request but currently it's only updating the source RSE and not the source URL which leads to issues in the monitoring data.

Modification
------------
Update the source URL correctly.



Changed files:
lib/rucio/tests/test_conveyor.py, lib/rucio/core/request.py, lib/rucio/transfertool/fts3.py, lib/rucio/daemons/conveyor/receiver.py

diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index 3c9d129f8..6cb588890 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -1248,7 +1248,7 @@ def update_request_state(response, session=None, logger=logging.log):
                 transfer_id = response['transfer_id'] if 'transfer_id' in response else None
                 logger(logging.INFO, 'UPDATING REQUEST %s FOR TRANSFER %s STATE %s' % (str(response['request_id']), transfer_id, str(response['new_state'])))
 
-                job_m_replica = response.get('job_m_replica', None)
+                multi_sources = response.get('multi_sources', None)
                 src_url = response.get('src_url', None)
                 src_rse = response.get('src_rse', None)
                 src_rse_id = response.get('src_rse_id', None)
@@ -1256,7 +1256,7 @@ def update_request_state(response, session=None, logger=logging.log):
                 staging_finished_at = response.get('staging_finished', None)
                 started_at = response.get('started_at', None)
                 transferred_at = response.get('transferred_at', None)
-                if job_m_replica and (str(job_m_replica).lower() == str('true')) and src_url:
+                if multi_sources and (str(multi_sources).lower() == str('true')) and src_url:
                     try:
                         src_rse_name, src_rse_id = __get_source_rse(response['request_id'], src_url, session=session)
                     except Exception:
diff --git a/lib/rucio/daemons/conveyor/receiver.py b/lib/rucio/daemons/conveyor/receiver.py
index d80cfb115..3d9811bd0 100644
--- a/lib/rucio/daemons/conveyor/receiver.py
+++ b/lib/rucio/daemons/conveyor/receiver.py
@@ -115,7 +115,7 @@ class Receiver(object):
                             'md5': msg['file_metadata'].get('md5', None),
                             'filesize': msg['file_metadata'].get('filesize', None),
                             'external_host': msg.get('endpnt', None),
-                            'job_m_replica': msg.get('job_m_replica', None),
+                            'multi_sources': msg.get('job_metadata', {}).get('multi_sources', None),
                             'details': {'files': msg['file_metadata']}}
 
                 record_counter('daemons.conveyor.receiver.message_rucio')
diff --git a/lib/rucio/tests/test_conveyor.py b/lib/rucio/tests/test_conveyor.py
index c92911af8..fe149c1ff 100644
--- a/lib/rucio/tests/test_conveyor.py
+++ b/lib/rucio/tests/test_conveyor.py
@@ -321,8 +321,20 @@ def test_multisource(vo, did_factory, root_account, replica_client, core_config_
     assert __source_exists(src_rse_id=src_rse1_id, **did)
     assert __source_exists(src_rse_id=src_rse2_id, **did)
 
+    # After submission, the source rse is the one which will fail
+    request = request_core.get_request_by_did(rse_id=dst_rse_id, **did)
+    assert request['source_rse'] == src_rse2
+    assert request['source_rse_id'] == src_rse2_id
+
+    # The source_rse must be updated to the correct one
+    request = __wait_for_request_state(dst_rse_id=dst_rse_id, state=RequestState.DONE, **did)
+    assert request['source_rse'] == src_rse1
+    assert request['source_rse_id'] == src_rse1_id
+
     replica = __wait_for_replica_transfer(dst_rse_id=dst_rse_id, **did)
     assert replica['state'] == ReplicaState.AVAILABLE
+
+    # Both entries in source table must be removed after completion
     assert not __source_exists(src_rse_id=src_rse1_id, **did)
     assert not __source_exists(src_rse_id=src_rse2_id, **did)
 
@@ -360,6 +372,11 @@ def test_multisource_receiver(vo, did_factory, replica_client, root_account, met
         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)
         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)
 
+        # After submission, the source rse is the one which will fail
+        request = request_core.get_request_by_did(rse_id=dst_rse_id, **did)
+        assert request['source_rse'] == src_rse2
+        assert request['source_rse_id'] == src_rse2_id
+
         request = None
         for _ in range(MAX_POLL_WAIT_SECONDS):
             request = request_core.get_request_by_did(rse_id=dst_rse_id, **did)
@@ -372,6 +389,9 @@ def test_multisource_receiver(vo, did_factory, replica_client, root_account, met
         assert request['state'] == RequestState.DONE
 
         assert metrics_mock.get_sample_value('rucio_daemons_conveyor_receiver_update_request_state_total', labels={'updated': 'True'}) >= 1
+        # The source was updated to the good one
+        assert request['source_rse'] == src_rse1
+        assert request['source_rse_id'] == src_rse1_id
     finally:
         receiver_graceful_stop.set()
         receiver_thread.join(timeout=5)
diff --git a/lib/rucio/transfertool/fts3.py b/lib/rucio/transfertool/fts3.py
index 2f60f9c90..b96a61258 100644
--- a/lib/rucio/transfertool/fts3.py
+++ b/lib/rucio/transfertool/fts3.py
@@ -682,7 +682,7 @@ class FTS3Transfertool(Transfertool):
                                  'md5': file_resp['file_metadata'].get('md5', None),
                                  'filesize': file_resp['file_metadata'].get('filesize', None),
                                  'external_host': self.external_host,
-                                 'job_m_replica': multi_sources,
+                                 'multi_sources': multi_sources,
                                  'details': {'files': file_resp['file_metadata']}}
 
             # multiple source replicas jobs and we found the successful one, it's the final state.