Transfers: correctly record updated requests on failed multihops

When handling a multi-hop transfer, all the hops following the one
which failed are automatically failed by the poller/receiver.
Until recently, those hops where handled again separately. However,
a recent change started ignoring requests which where already handled.
As a result, the metrics don't record correctly the number of
updated requests.

lib/rucio/core/request.py

@@ -1096,9 +1096,10 @@ def is_intermediate_hop(request):
 
 
 @transactional_session
-def handle_failed_intermediate_hop(request, *, session: "Session"):
+def handle_failed_intermediate_hop(request, *, session: "Session") -> int:
     """
     Perform housekeeping behind a failed intermediate hop
+    Returns the number of updated requests
     """
     # mark all hops following this one (in any multihop path) as Failed
     new_state = RequestState.FAILED
@@ -1123,6 +1124,7 @@ def handle_failed_intermediate_hop(request, *, session: "Session"):
             err_msg=get_transfer_error(new_state, reason=reason),
         )
         session.execute(stmt)
+    return len(dependent_requests)
 
 
 @METRICS.count_it

lib/rucio/core/transfer.py

@@ -540,10 +540,11 @@ def update_transfer_state(
     :param tt_status_report:      The transfertool status update, retrieved via request.query_request().
     :param session:               The database session to use.
     :param logger:                Optional decorated logger that can be passed from the calling daemons or servers.
-    :returns commit_or_rollback:  Boolean.
+    :returns:                     The number of updated requests
     """
 
     request_id = tt_status_report.request_id
+    nb_updated = 0
     try:
         fields_to_update = tt_status_report.get_db_fields_to_update(session=session, logger=logger)
         if not fields_to_update:
@@ -556,11 +557,12 @@ def update_transfer_state(
             updated = transition_request_state(request_id, request=request, session=session, **fields_to_update)
 
             if not updated:
-                return False
+                return nb_updated
+            nb_updated += 1
 
             if tt_status_report.state == RequestState.FAILED:
                 if request_core.is_intermediate_hop(request):
-                    request_core.handle_failed_intermediate_hop(request, session=session)
+                    nb_updated += request_core.handle_failed_intermediate_hop(request, session=session)
 
             if tt_status_report.state:
                 stats_manager.observe(
@@ -580,10 +582,10 @@ def update_transfer_state(
                 additional_fields=tt_status_report.get_monitor_msg_fields(session=session, logger=logger),
                 session=session
             )
-            return True
+            return nb_updated
     except UnsupportedOperation as error:
         logger(logging.WARNING, "Request %s doesn't exist - Error: %s" % (request_id, str(error).replace('\n', '')))
-        return False
+        return 0
     except Exception:
         logger(logging.CRITICAL, "Exception", exc_info=True)
 

lib/rucio/daemons/conveyor/poller.py

@@ -379,10 +379,11 @@ def _poll_transfers(
                         stats_manager=transfer_stats_manager,
                         logger=logger,
                     )
-                    # if True, really update request content; if False, only touch request
+                    cnt += ret
                     if ret:
-                        cnt += 1
-                    METRICS.counter('update_request_state.{updated}').labels(updated=ret).inc()
+                        METRICS.counter('update_request_state.{updated}').labels(updated=True).inc(delta=ret)
+                    else:
+                        METRICS.counter('update_request_state.{updated}').labels(updated=False).inc()
 
             # should touch transfers.
             # Otherwise if one bulk transfer includes many requests and one is not terminated, the transfer will be poll again.

lib/rucio/daemons/conveyor/receiver.py

@@ -93,7 +93,10 @@ def _perform_request_update(self, msg, *, session=None, logger=logging.log):
                     session=session,
                     logger=logger,
                 )
-                METRICS.counter('update_request_state.{updated}').labels(updated=ret).inc()
+                if ret:
+                    METRICS.counter('update_request_state.{updated}').labels(updated=True).inc(delta=ret)
+                else:
+                    METRICS.counter('update_request_state.{updated}').labels(updated=False).inc()
         except Exception:
             logging.critical(traceback.format_exc())