Fix data store bug where late arrival of submitted message would cause backwards state change (#6914)

Fix data store bug where late arrival of submitted message would cause backwards state change

Author: MetRonnie

changes.d/6914.fix.md

@@ -0,0 +1 @@
+Fixed bug where jobs in the UI could regress to an earlier state.

cylc/flow/data_store_mgr.py

@@ -67,6 +67,7 @@
     Set,
     TYPE_CHECKING,
     Tuple,
+    Union,
 )
 import zlib
 
@@ -1609,13 +1610,20 @@ def _apply_broadcasts_to_runtime(self, tokens, rtconfig):
             poverride(rtconfig, overrides, prepend=True)
         return rtconfig
 
-    def insert_job(self, name, cycle_point, status, job_conf):
+    def insert_job(
+        self,
+        name: str,
+        cycle_point: Union['PointBase', str],
+        status: str,
+        job_conf: dict,
+    ):
         """Insert job into data-store.
 
         Args:
-            name (str): Corresponding task name.
-            cycle_point (str|PointBase): Cycle point string
-            job_conf (dic):
+            name: Corresponding task name.
+            cycle_point: Cycle point string
+            status: The task's state.
+            job_conf:
                 Dictionary of job configuration used to generate
                 the job script.
                 (see TaskJobManager._prep_submit_task_job_impl)
@@ -1625,6 +1633,11 @@ def insert_job(self, name, cycle_point, status, job_conf):
             None
 
         """
+        if status not in JOB_STATUS_SET:
+            # Ignore task-only states e.g. preparing
+            # https://github.com/cylc/cylc-flow/issues/4994
+            return
+
         sub_num = job_conf['submit_num']
         tp_tokens = self.id_.duplicate(
             cycle=str(cycle_point),
@@ -1641,9 +1654,6 @@ def insert_job(self, name, cycle_point, status, job_conf):
             # Job already exists (i.e. post-submission submit failure)
             return
 
-        if status not in JOB_STATUS_SET:
-            return
-
         j_buf = PbJob(
             stamp=f'{j_id}@{update_time}',
             id=j_id,
@@ -2761,12 +2771,12 @@ def delta_job_msg(self, tokens: Tokens, msg: str) -> None:
 
     def delta_job_attr(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         attr_key: str,
         attr_val: Any,
     ) -> None:
         """Set job attribute."""
-        j_id, job = self.store_node_fetcher(tokens)
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
         if not job:
             return
         j_delta = PbJob(stamp=f'{j_id}@{time()}')
@@ -2779,12 +2789,19 @@ def delta_job_attr(
 
     def delta_job_state(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         status: str,
     ) -> None:
         """Set job state."""
-        j_id, job = self.store_node_fetcher(tokens)
-        if not job or status not in JOB_STATUS_SET:
+        if status not in JOB_STATUS_SET:
+            # Ignore task-only states e.g. preparing
+            return
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
+        if not job or (
+            # Don't cause backwards state change:
+            JOB_STATUSES_ALL.index(status) <= JOB_STATUSES_ALL.index(job.state)
+            and not itask.job_vacated
+        ):
             return
         j_delta = PbJob(
             stamp=f'{j_id}@{time()}',
@@ -2798,15 +2815,15 @@ def delta_job_state(
 
     def delta_job_time(
         self,
-        tokens: Tokens,
+        itask: 'TaskProxy',
         event_key: str,
         time_str: Optional[str] = None,
     ) -> None:
         """Set an event time in job pool object.
 
         Set values of both event_key + '_time' and event_key + '_time_string'.
         """
-        j_id, job = self.store_node_fetcher(tokens)
+        j_id, job = self.store_node_fetcher(itask.job_tokens)
         if not job:
             return
         j_delta = PbJob(stamp=f'{j_id}@{time()}')

cylc/flow/task_events_mgr.py

@@ -797,11 +797,9 @@ def process_message(
             # ... but either way update the job ID in the job proxy (it only
             # comes in via the submission message).
             if itask.run_mode != RunMode.SIMULATION:
-                job_tokens = itask.tokens.duplicate(
-                    job=str(itask.submit_num)
-                )
                 self.data_store_mgr.delta_job_attr(
-                    job_tokens, 'job_id', itask.summary['submit_method_id'])
+                    itask, 'job_id', itask.summary['submit_method_id']
+                )
             else:
                 # In simulation mode submitted implies started:
                 self.spawn_children(itask, TASK_OUTPUT_STARTED, forced)
@@ -1354,9 +1352,8 @@ def _process_message_failed(
         if event_time is None:
             event_time = get_current_time_string()
         itask.set_summary_time('finished', event_time)
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'finished', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_FAILED)
+        self.data_store_mgr.delta_job_time(itask, 'finished', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_FAILED)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "run_status": 1,
             "time_run_exit": event_time,
@@ -1394,12 +1391,11 @@ def _process_message_failed(
 
     def _process_message_started(self, itask, event_time, forced):
         """Helper for process_message, handle a started message."""
+        self.data_store_mgr.delta_job_time(itask, 'started', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_RUNNING)
         if itask.job_vacated:
             itask.job_vacated = False
             LOG.info(f"[{itask}] Vacated job restarted")
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'started', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_RUNNING)
         itask.set_summary_time('started', event_time)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "time_run": itask.summary['started_time_string']})
@@ -1430,9 +1426,8 @@ def _process_message_succeeded(self, itask, event_time, forced):
         Ignore forced.
         """
 
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(job_tokens, 'finished', event_time)
-        self.data_store_mgr.delta_job_state(job_tokens, TASK_STATUS_SUCCEEDED)
+        self.data_store_mgr.delta_job_time(itask, 'finished', event_time)
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUCCEEDED)
         itask.set_summary_time('finished', event_time)
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "run_status": 0,
@@ -1505,13 +1500,9 @@ def _process_message_submit_failed(
             self.setup_event_handlers(itask, self.EVENT_SUBMIT_RETRY, msg)
 
         # Register newly submit-failed job with the database and datastore.
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
         self._insert_task_job(
             itask, event_time, self.JOB_SUBMIT_FAIL_FLAG, forced=forced)
-        self.data_store_mgr.delta_job_state(
-            job_tokens,
-            TASK_STATUS_SUBMIT_FAILED
-        )
+        self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUBMIT_FAILED)
         self._reset_job_timers(itask)
 
         return no_retries
@@ -1560,24 +1551,12 @@ def _process_message_submitted(
         # Do after itask has changed state
         self._insert_task_job(
             itask, event_time, self.JOB_SUBMIT_SUCCESS_FLAG, forced=forced)
-        job_tokens = itask.tokens.duplicate(job=str(itask.submit_num))
-        self.data_store_mgr.delta_job_time(
-            job_tokens,
-            'submitted',
-            event_time,
-        )
+        self.data_store_mgr.delta_job_time(itask, 'submitted', event_time)
         if itask.run_mode == RunMode.SIMULATION:
             # Simulate job started as well.
-            self.data_store_mgr.delta_job_time(
-                job_tokens,
-                'started',
-                event_time,
-            )
+            self.data_store_mgr.delta_job_time(itask, 'started', event_time)
         else:
-            self.data_store_mgr.delta_job_state(
-                job_tokens,
-                TASK_STATUS_SUBMITTED,
-            )
+            self.data_store_mgr.delta_job_state(itask, TASK_STATUS_SUBMITTED)
 
     def _insert_task_job(
         self,

cylc/flow/task_proxy.py

@@ -315,6 +315,11 @@ def __init__(
                 )
             )
 
+    @property
+    def job_tokens(self) -> 'Tokens':
+        """Return the job tokens for this task proxy."""
+        return self.tokens.duplicate(job=str(self.submit_num))
+
     def __repr__(self) -> str:
         return f"<{type(self).__name__} {self.identity} {self.state}>"