Repository: rucio/rucio
Issue: requests stay forever in Queued state if source creation fails

Motivation
----------
If an exception occurs while trying to create sources on submission, the database transaction is rollbacked and the requests remain
in queued state. 


Modification
------------

The requests should be transited to SUBMISSION_FAILED state. 



Changed files:
lib/rucio/core/transfer.py, lib/rucio/daemons/conveyor/common.py

diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index f7b32a804..291fc745a 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -480,64 +480,57 @@ def transfer_path_str(transfer_path: "List[DirectTransferDefinition]") -> str:
 
 
 @transactional_session
-def mark_submitting_and_prepare_sources_for_transfers(
-        transfers: "Iterable[DirectTransferDefinition]",
+def mark_submitting_and_prepare_sources_for_transfer(
+        transfer: "DirectTransferDefinition",
         external_host: str,
         logger: "Callable",
         session: "Optional[Session]" = None,
 ):
     """
     Prepare the sources for transfers.
-    :param transfers:  Dictionary containing request transfer info.
+    :param transfer:   A transfer object
     :param session:    Database session to use.
     """
 
-    logger(logging.DEBUG, 'Start to prepare transfer')
-    try:
-        for transfer in transfers:
-            log_str = 'PREPARING REQUEST %s DID %s:%s TO SUBMITTING STATE PREVIOUS %s FROM %s TO %s USING %s ' % (transfer.rws.request_id,
-                                                                                                                  transfer.rws.scope,
-                                                                                                                  transfer.rws.name,
-                                                                                                                  transfer.rws.previous_attempt_id,
-                                                                                                                  transfer.legacy_sources,
-                                                                                                                  transfer.dest_url,
-                                                                                                                  external_host)
-            logger(logging.INFO, "%s", log_str)
-
-            rowcount = session.query(models.Request)\\
-                              .filter_by(id=transfer.rws.request_id)\\
-                              .filter(models.Request.state == RequestState.QUEUED)\\
-                              .update({'state': RequestState.SUBMITTING,
-                                       'external_id': None,
-                                       'external_host': external_host,
-                                       'dest_url': transfer.dest_url,
-                                       'submitted_at': datetime.datetime.utcnow()},
-                                      synchronize_session=False)
-            if rowcount == 0:
-                raise RequestNotFound("Failed to prepare transfer: request %s does not exist or is not in queued state" % transfer.rws)
-
-            for src_rse, src_url, src_rse_id, rank in transfer.legacy_sources:
-                # For multi-hops, sources in database are bound to the initial request
-                source_request_id = transfer.rws.attributes.get('initial_request_id', transfer.rws.request_id)
-                src_rowcount = session.query(models.Source)\\
-                                      .filter_by(request_id=source_request_id)\\
-                                      .filter(models.Source.rse_id == src_rse_id)\\
-                                      .update({'is_using': True}, synchronize_session=False)
-                if src_rowcount == 0:
-                    models.Source(request_id=source_request_id,
-                                  scope=transfer.rws.scope,
-                                  name=transfer.rws.name,
-                                  rse_id=src_rse_id,
-                                  dest_rse_id=transfer.dst.rse.id,
-                                  ranking=rank if rank else 0,
-                                  bytes=transfer.rws.byte_count,
-                                  url=src_url,
-                                  is_using=True).\\
-                        save(session=session, flush=False)
-
-    except IntegrityError as error:
-        raise RucioException(error.args)
-    logger(logging.DEBUG, 'Finished to prepare transfer')
+    log_str = 'PREPARING REQUEST %s DID %s:%s TO SUBMITTING STATE PREVIOUS %s FROM %s TO %s USING %s ' % (transfer.rws.request_id,
+                                                                                                          transfer.rws.scope,
+                                                                                                          transfer.rws.name,
+                                                                                                          transfer.rws.previous_attempt_id,
+                                                                                                          transfer.legacy_sources,
+                                                                                                          transfer.dest_url,
+                                                                                                          external_host)
+    logger(logging.INFO, "%s", log_str)
+
+    rowcount = session.query(models.Request)\\
+                      .filter_by(id=transfer.rws.request_id)\\
+                      .filter(models.Request.state == RequestState.QUEUED)\\
+                      .update({'state': RequestState.SUBMITTING,
+                               'external_id': None,
+                               'external_host': external_host,
+                               'dest_url': transfer.dest_url,
+                               'submitted_at': datetime.datetime.utcnow()},
+                              synchronize_session=False)
+    if rowcount == 0:
+        raise RequestNotFound("Failed to prepare transfer: request %s does not exist or is not in queued state" % transfer.rws)
+
+    for src_rse, src_url, src_rse_id, rank in transfer.legacy_sources:
+        # For multi-hops, sources in database are bound to the initial request
+        source_request_id = transfer.rws.attributes.get('initial_request_id', transfer.rws.request_id)
+        src_rowcount = session.query(models.Source)\\
+                              .filter_by(request_id=source_request_id)\\
+                              .filter(models.Source.rse_id == src_rse_id)\\
+                              .update({'is_using': True}, synchronize_session=False)
+        if src_rowcount == 0:
+            models.Source(request_id=source_request_id,
+                          scope=transfer.rws.scope,
+                          name=transfer.rws.name,
+                          rse_id=src_rse_id,
+                          dest_rse_id=transfer.dst.rse.id,
+                          ranking=rank if rank else 0,
+                          bytes=transfer.rws.byte_count,
+                          url=src_url,
+                          is_using=True).\\
+                save(session=session, flush=False)
 
 
 @transactional_session
diff --git a/lib/rucio/daemons/conveyor/common.py b/lib/rucio/daemons/conveyor/common.py
index 08d5499b8..fec5a223e 100644
--- a/lib/rucio/daemons/conveyor/common.py
+++ b/lib/rucio/daemons/conveyor/common.py
@@ -54,6 +54,7 @@ from rucio.common.exception import (InvalidRSEExpression, TransferToolTimeout, T
 from rucio.common.logging import formatted_logger
 from rucio.core import heartbeat, request, transfer as transfer_core
 from rucio.core.monitor import record_counter, record_timer
+from rucio.core.request import set_request_state
 from rucio.core.rse import list_rses
 from rucio.core.rse_expression_parser import parse_expression
 from rucio.core.vo import list_vos
@@ -117,14 +118,16 @@ def submit_transfer(transfertool_obj, transfers, job_params, submitter='submitte
     :param logger:                Optional decorated logger that can be passed from the calling daemons or servers.
     """
 
-    try:
-        transfer_core.mark_submitting_and_prepare_sources_for_transfers(transfers, external_host=transfertool_obj.external_host, logger=logger)
-    except RequestNotFound as error:
-        logger(logging.ERROR, str(error))
-        return
-    except Exception:
-        logger(logging.ERROR, 'Failed to prepare requests %s state to SUBMITTING (Will not submit jobs but return directly) with error' % [str(t.rws) for t in transfers], exc_info=True)
-        return
+    for transfer in transfers:
+        try:
+            transfer_core.mark_submitting_and_prepare_sources_for_transfer(transfer, external_host=transfertool_obj.external_host, logger=logger)
+        except RequestNotFound as error:
+            logger(logging.ERROR, str(error))
+            return
+        except Exception:
+            logger(logging.ERROR, 'Failed to prepare requests %s state to SUBMITTING. Mark it SUBMISSION_FAILED and abort submission.' % [str(t.rws) for t in transfers], exc_info=True)
+            set_request_state(request_id=transfer.rws.request_id, new_state=RequestState.SUBMISSION_FAILED)
+            return
 
     try:
         _submit_transfers(transfertool_obj, transfers, job_params, submitter, timeout, logger)