[9.1] fix (WMS): stop the StalledJobAgent rescheduling jobs that have just been matched

src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py

@@ -561,34 +561,39 @@ def _kickStuckJobs(self):
         """Reschedule jobs stuck in initialization status Rescheduled, Matched."""
 
         message = ""
-
-        checkTime = DiracTime.utcnow() - self.matchedTime * second
-        result = self.jobDB.selectJobs({"Status": JobStatus.MATCHED}, older=checkTime)
-        if not result["OK"]:
-            self.log.error("Failed to select jobs", result["Message"])
-            return result
-
-        jobIDs = result["Value"]
-        if jobIDs:
-            self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in {JobStatus.MATCHED} status")
-            result = rescheduleJobs(jobIDs)
+        for status, stuckTime in ((JobStatus.MATCHED, self.matchedTime), (JobStatus.RESCHEDULED, self.rescheduledTime)):
+            checkTime = DiracTime.utcnow() - stuckTime * second
+            result = self.jobDB.selectJobs({"Status": status}, older=checkTime)
             if not result["OK"]:
-                message = f"Failed to reschedule jobs stuck in {JobStatus.MATCHED} status"
-                message += "\n" + result["Message"]
-
-        checkTime = DiracTime.utcnow() - self.rescheduledTime * second
-        result = self.jobDB.selectJobs({"Status": JobStatus.RESCHEDULED}, older=checkTime)
-        if not result["OK"]:
-            self.log.error("Failed to select jobs", result["Message"])
-            return result
+                self.log.error("Failed to select jobs", result["Message"])
+                return result
 
-        jobIDs = result["Value"]
-        if jobIDs:
-            self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in {JobStatus.RESCHEDULED} status")
-            result = rescheduleJobs(jobIDs)
-            if not result["OK"]:
-                message = f"Failed to reschedule jobs stuck in {JobStatus.RESCHEDULED} status"
-                message += "\n" + result["Message"]
+            jobIDs = result["Value"]
+            if not jobIDs:
+                continue
+            self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in {status} status")
+            failedJobIDs = []
+            for jobID in jobIDs:
+                # The job may have moved on since it was selected (e.g. a pilot picked it up),
+                # so check again just before rescheduling it
+                result = self.jobDB.selectJobs({"JobID": jobID, "Status": status}, older=checkTime)
+                if not result["OK"]:
+                    failedJobIDs.append(jobID)
+                    continue
+                if not result["Value"]:
+                    self.log.info("Job is no longer stuck, not rescheduling", f"{jobID} ({status})")
+                    continue
+                result = rescheduleJobs(
+                    [jobID],
+                    source="StalledJobAgent",
+                    jobDB=self.jobDB,
+                    taskQueueDB=self.taskQueueDB,
+                    jobLoggingDB=self.logDB,
+                )
+                if not result["OK"]:
+                    failedJobIDs.append(jobID)
+            if failedJobIDs:
+                message += f"Failed to reschedule jobs stuck in {status} status: {failedJobIDs}\n"
 
         if message:
             self.log.error(message)

src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py

@@ -72,3 +72,25 @@ def test__sjaFunctions(sja):
     assert sja._kickStuckJobs()["OK"]
     assert sja._failStalledJobs(0)["OK"]
     assert not sja._markStalledJobs(0)["OK"]
+
+
+def test__kickStuckJobs_recheck(sja, mocker):
+    """Jobs that move on between the selection and the rescheduling are left alone"""
+    stillStuck = {1}
+
+    def selectJobs(condDict, older=None):
+        if "JobID" in condDict:
+            return {"OK": True, "Value": [condDict["JobID"]] if condDict["JobID"] in stillStuck else []}
+        if condDict["Status"] == "Matched":
+            return {"OK": True, "Value": [1, 2]}
+        return {"OK": True, "Value": []}
+
+    sja.jobDB.selectJobs.side_effect = selectJobs
+    reschedule = mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.rescheduleJobs", return_value={"OK": True}
+    )
+
+    assert sja._kickStuckJobs()["OK"]
+    reschedule.assert_called_once()
+    assert reschedule.call_args.args == ([1],)
+    assert reschedule.call_args.kwargs["source"] == "StalledJobAgent"

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -405,16 +405,12 @@ def setJobAttribute(self, jobID, attrName, attrValue, update=False, myDate=None,
 
         if attrName == "Status":
             # Treat this update separately
-            res = self.setJobsMajorStatus([jobID], attrValue, force=force)
+            res = self.setJobsMajorStatus([jobID], attrValue, force=force, update=update and not myDate)
             if not res["OK"]:
                 return res
-            if update:
-                cmd = "UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID=%s"
-                args = [str(jobID)]
-                if myDate:
-                    cmd += " AND LastUpdateTime < %s"
-                    args.append(myDate)
-                return self._update(cmd, args=args)
+            if update and myDate:
+                cmd = "UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID=%s AND LastUpdateTime < %s"
+                return self._update(cmd, args=[str(jobID), myDate])
             else:
                 return res
 
@@ -474,7 +470,9 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No
         statusDone = False
         if "Status" in attrNames:
             # Treat this update separately
-            res = self.setJobsMajorStatus(jIDList, attrValues[attrNames.index("Status")], force=force)
+            res = self.setJobsMajorStatus(
+                jIDList, attrValues[attrNames.index("Status")], force=force, update=update and not myDate
+            )
             if not res["OK"]:
                 return res
             attrValues.pop(attrNames.index("Status"))
@@ -507,12 +505,14 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No
 
         return self._update(cmd, args=args)
 
-    def setJobsMajorStatus(self, jIDList, candidateStatus, force=False):
+    def setJobsMajorStatus(self, jIDList, candidateStatus, force=False, update=False):
         """
         Sets jobs major status, considering the JobStateMachine result
 
         :param list jIDList: list of one or more job IDs
         :param str candidateStatus: candidate major Status
+        :param bool update: also refresh the LastUpdateTime stamp in the same statement, so that
+                            the new status is never visible together with the old time stamp
         """
 
         # get the current statuses of the jobs
@@ -543,6 +543,8 @@ def setJobsMajorStatus(self, jIDList, candidateStatus, force=False):
             newStatuses[jID] = nextState
 
         cmd = "INSERT INTO Jobs (JobID, Status) VALUES (%s, %s) ON DUPLICATE KEY UPDATE Status=VALUES(Status)"
+        if update:
+            cmd += ", LastUpdateTime=UTC_TIMESTAMP()"
         return self._updatemany(cmd, data=newStatuses.items())
 
     def setJobStatus(self, jobID, status="", minorStatus="", applicationStatus="", force=False):