From 9c7ea5ddae3db3a8ddc168c6204c73aa731ffe1c 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. --- src/DIRAC/WorkloadManagementSystem/DB/JobDB.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index 39354ec45f4..fe5719f681a 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py @@ -467,13 +467,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: + if update and myDate: cmd = f"UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID={jobID}" - if myDate: - cmd += f" AND LastUpdateTime < {myDate}" + cmd += f" AND LastUpdateTime < {myDate}" return self._update(cmd) else: return res @@ -542,7 +541,9 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No 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")) @@ -567,12 +568,14 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No return self._update(cmd) - 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 @@ -614,6 +617,8 @@ def setJobsMajorStatus(self, jIDList, candidateStatus, force=False): cmd += ",".join(ns) cmd += " ON DUPLICATE KEY UPDATE Status=VALUES(Status)" + if update: + cmd += ", LastUpdateTime=UTC_TIMESTAMP()" return self._update(cmd) From 9fdc44285f4848227d11c94be957e3a2ccf00add 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. --- .../Agent/StalledJobAgent.py | 50 +++++++++---------- .../Agent/test/Test_Agent_StalledJobAgent.py | 18 +++++++ 2 files changed, 43 insertions(+), 25 deletions(-) diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py index ca6766fcc42..b849eecf08d 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py @@ -515,34 +515,34 @@ def _kickStuckJobs(self): """Reschedule jobs stuck in initialization status Rescheduled, Matched""" message = "" + for status, stuckTime in ((JobStatus.MATCHED, self.matchedTime), (JobStatus.RESCHEDULED, self.rescheduledTime)): + checkTime = datetime.datetime.utcnow() - stuckTime * second + result = self.jobDB.selectJobs({"Status": status}, older=checkTime) + if not result["OK"]: + self.log.error("Failed to select jobs", result["Message"]) + return result - checkTime = datetime.datetime.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 Matched status") - result = self.jobDB.rescheduleJobs(jobIDs) - if "FailedJobs" in result: - message = f"Failed to reschedule {len(result['FailedJobs'])} jobs stuck in Matched status" - - checkTime = datetime.datetime.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 - - jobIDs = result["Value"] - if jobIDs: - self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in Rescheduled status") - result = self.jobDB.rescheduleJobs(jobIDs) - if "FailedJobs" in result: + 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 + if not self.jobDB.rescheduleJob(jobID)["OK"]: + failedJobIDs.append(jobID) + if failedJobIDs: if message: message += "\n" - message += f"Failed to reschedule {len(result['FailedJobs'])} jobs stuck in Rescheduled status" + message += f"Failed to reschedule {len(failedJobIDs)} jobs stuck in {status} status" if message: return S_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 dbf4ce2fbf7..382cc5baeb1 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py @@ -49,3 +49,21 @@ 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 + sja.jobDB.rescheduleJob.return_value = {"OK": True} + + assert sja._kickStuckJobs()["OK"] + sja.jobDB.rescheduleJob.assert_called_once_with(1)