From 16db1233c1e77c9ea7b6748b0cafef00d51d1fc7 Mon Sep 17 00:00:00 2001 From: Chris Burr Date: Fri, 23 Oct 2026 00:41:18 +0200 Subject: [PATCH 1/2] fix (WMS): set the job Status and LastUpdateTime in a single statement When setJobAttributes is called with update=True and a Status, the status and the LastUpdateTime were written by two separate statements. In between, a freshly matched job that had been waiting for longer than MatchedTime looked like a job stuck in Matched, and the StalledJobAgent could reschedule it while the pilot went on to run it. The job then stays in Checking forever, as the state machine rejects the pilot's updates. --- .../WorkloadManagementSystem/DB/JobDB.py | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index b5e8cb005ad..97056e6972e 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/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): From 66dfb261d4a559952c809dece7802eb43b95062e Mon Sep 17 00:00:00 2001 From: Chris Burr Date: Fri, 23 Oct 2026 00:41:19 +0200 Subject: [PATCH 2/2] fix (WMS): StalledJobAgent re-checks stuck jobs before rescheduling them Jobs selected as stuck in Matched or Rescheduled may have moved on by the time they are rescheduled, so check each one again just before acting. Also record the StalledJobAgent as the source of the reschedule in the job logging. --- .../Agent/StalledJobAgent.py | 57 ++++++++++--------- .../Agent/test/Test_Agent_StalledJobAgent.py | 22 +++++++ 2 files changed, 53 insertions(+), 26 deletions(-) diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py index 5b33f31feaa..60889323714 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py +++ b/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) diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py index 0f49936cdb2..b2481844df9 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py +++ b/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"