Merge pull request #7847 from chrisburr/fix-fts3-deadlock

[rel-v8r0] Fix deadlock when FTS3Agent._treatOperation fails
DIRACGrid · Oct 23, 2024 · 4c0a23d · 4c0a23d
2 parents 56268a1 + 475bddb
commit 4c0a23d
Showing 1 changed file with 17 additions and 13 deletions.
diff --git a/src/DIRAC/DataManagementSystem/Agent/FTS3Agent.py b/src/DIRAC/DataManagementSystem/Agent/FTS3Agent.py
@@ -288,13 +288,15 @@ def _monitorJobCallback(returnedValue):
         :param returnedValue: value returned by the _monitorJob method
                               (ftsJob, standard dirac return struct)
         """
-
-        ftsJob, res = returnedValue
-        log = gLogger.getLocalSubLogger(f"_monitorJobCallback/{ftsJob.jobID}")
-        if not res["OK"]:
-            log.error("Error updating job status", res)
+        if not isinstance(returnedValue, tuple) or len(returnedValue) != 2:
+            ftsJob, res = returnedValue
+            log = gLogger.getLocalSubLogger(f"_monitorJobCallback/{ftsJob.jobID}")
+            if not res["OK"]:
+                log.error("Error updating job status", res)
+            else:
+                log.debug("Successfully updated job status")
         else:
-            log.debug("Successfully updated job status")
+            log.error("Invalid return value when monitoring job", f"{returnedValue!r}")
 
     def monitorJobsLoop(self):
         """* fetch the active FTSJobs from the DB
@@ -372,13 +374,15 @@ def _treatOperationCallback(returnedValue):
         :param returnedValue: value returned by the _treatOperation method
                               (ftsOperation, standard dirac return struct)
         """
-
-        operation, res = returnedValue
-        log = gLogger.getLocalSubLogger(f"_treatOperationCallback/{operation.operationID}")
-        if not res["OK"]:
-            log.error("Error treating operation", res)
+        if isinstance(returnedValue, tuple) and len(returnedValue) == 2:
+            operation, res = returnedValue
+            log = gLogger.getLocalSubLogger(f"_treatOperationCallback/{operation.operationID}")
+            if not res["OK"]:
+                log.error("Error treating operation", res)
+            else:
+                log.debug("Successfully treated operation")
         else:
-            log.debug("Successfully treated operation")
+            log.error("Invalid return value when treating operation", f"{returnedValue!r}")
 
     def _treatOperation(self, operation):
         """Treat one operation:
@@ -510,7 +514,7 @@ def _treatOperation(self, operation):
                                 scope=["fts"],
                             )
                             if not res["OK"]:
-                                return res
+                                return operation, res
 
                             fts_access_token = res["Value"]["access_token"]