Merge pull request #6842 from MetRonnie/cylc-set

Fix `cylc set` causing erroneous state changes

Author: hjoliver

changes.d/6842.fix.md

@@ -0,0 +1,3 @@
+Fixed two `cylc set` bugs:
+- It was affecting job states when it should only affect _task_ states.
+- In some cases it was causing a task to go back into the waiting state.

cylc/flow/data_store_mgr.py

@@ -134,6 +134,7 @@
     from cylc.flow.flow_mgr import FlowNums
     from cylc.flow.prerequisite import Prerequisite
     from cylc.flow.scheduler import Scheduler
+    from cylc.flow.taskdef import TaskDef
 
 EDGES = 'edges'
 FAMILIES = 'families'
@@ -256,7 +257,7 @@ def generate_checksum(in_strings):
     return zlib.adler32(''.join(sorted(in_strings)).encode()) & 0xffffffff
 
 
-def task_mean_elapsed_time(tdef):
+def task_mean_elapsed_time(tdef: 'TaskDef') -> float | None:
     """Calculate task mean elapsed time."""
     if tdef.elapsed_times:
         return round(sum(tdef.elapsed_times) / len(tdef.elapsed_times))
@@ -2739,11 +2740,14 @@ def delta_from_task_proxy(self, itask: TaskProxy) -> None:
     # -----------
     # Job Deltas
     # -----------
-    def delta_job_msg(self, tokens: Tokens, msg: str) -> None:
-        """Add message to job."""
+    def delta_job_msg(self, tokens: Tokens, msg: str) -> bool:
+        """Add message to job.
+
+        Returns False if the job was not found in the data store.
+        """
         j_id, job = self.store_node_fetcher(tokens)
         if not job:
-            return
+            return False
         j_delta = self.updated[JOBS].setdefault(
             j_id,
             PbJob(id=j_id)
@@ -2756,6 +2760,7 @@ def delta_job_msg(self, tokens: Tokens, msg: str) -> None:
             j_delta.messages[:] = job.messages
             j_delta.messages.append(msg)
         self.updates_pending = True
+        return True
 
     def delta_job_attr(
         self,

cylc/flow/id.py

@@ -23,6 +23,7 @@
 import re
 from typing import (
     TYPE_CHECKING,
+    Any,
     Iterable,
     List,
     Literal,
@@ -365,9 +366,22 @@ def is_null(self) -> bool:
             self[key] for key in self._REGULAR_KEYS
         )
 
+    @property
+    def submit_num(self) -> int | None:
+        """The job submit number as an integer, or None if not set.
+
+        Examples:
+            >>> Tokens('//c/t/01').submit_num
+            1
+            >>> Tokens('//c/t').submit_num is None
+            True
+        """
+        return int(self['job']) if self['job'] else None
+
     @overload
     def duplicate(
         self,
+        *,
         cycle: str,
         task: str,
         **kwargs,
@@ -377,13 +391,14 @@ def duplicate(
     @overload
     def duplicate(
         self,
+        *tokens_list: 'Tokens',
         **kwargs,
     ) -> 'Tokens':
         ...
 
     def duplicate(
         self,
-        *tokens_list,
+        *tokens_list: 'Tokens',
         **kwargs,
     ) -> 'Tokens':
         """Duplicate a tokens object.
@@ -418,7 +433,7 @@ def duplicate(
             '~u/w//c/b/01'
 
         """
-        _kwargs = {}
+        _kwargs: dict[str, Any] = {}
         for tokens in (self, *tokens_list):
             _kwargs.update(tokens)
         _kwargs.update(kwargs)

cylc/flow/network/resolvers.py

@@ -32,7 +32,6 @@
     Optional,
     Tuple,
     TYPE_CHECKING,
-    Union,
     cast,
 )
 from uuid import uuid4
@@ -69,9 +68,9 @@
 
 class TaskMsg(NamedTuple):
     """Tuple for Scheduler.message_queue"""
-    job_id: Union[Tokens, str]
+    job_id: Tokens
     event_time: str
-    severity: Union[str, int]
+    severity: str | int
     message: str
 
 
@@ -839,7 +838,12 @@ def put_messages(
         """
         for severity, message in messages:
             self.schd.message_queue.put(
-                TaskMsg(task_job, event_time, severity, message)
+                TaskMsg(
+                    Tokens(task_job, relative=True),
+                    event_time,
+                    severity,
+                    message,
+                )
             )
         return (True, f'Messages queued: {len(messages)}')
 

cylc/flow/run_modes/simulation.py

@@ -37,12 +37,16 @@
 from cylc.flow.exceptions import PointParsingError
 from cylc.flow.platforms import FORBIDDEN_WITH_PLATFORM
 from cylc.flow.run_modes import RunMode
-from cylc.flow.task_outputs import TASK_OUTPUT_SUBMITTED
+from cylc.flow.task_outputs import (
+    TASK_OUTPUT_STARTED,
+    TASK_OUTPUT_SUBMITTED,
+)
 from cylc.flow.task_state import (
     TASK_STATUS_FAILED,
     TASK_STATUS_RUNNING,
     TASK_STATUS_SUCCEEDED,
 )
+from cylc.flow.util import serialise_set
 from cylc.flow.wallclock import get_unix_time_from_time_string
 
 
@@ -91,22 +95,20 @@ def submit_task_job(
     itask.jobs.append(
         task_job_mgr.get_simulation_job_conf(itask)
     )
-    task_job_mgr.task_events_mgr.process_message(
-        itask, INFO, TASK_OUTPUT_SUBMITTED,
-    )
+    for output in (TASK_OUTPUT_SUBMITTED, TASK_OUTPUT_STARTED):
+        task_job_mgr.task_events_mgr.process_message(itask, INFO, output)
     task_job_mgr.workflow_db_mgr.put_insert_task_jobs(
         itask, {
             'time_submit': now[1],
             'time_run': now[1],
             'try_num': itask.get_try_num(),
-            'flow_nums': str(list(itask.flow_nums)),
+            'flow_nums': serialise_set(itask.flow_nums),
             'is_manual_submit': itask.is_manual_submit,
             'job_runner_name': RunMode.SIMULATION.value,
             'platform_name': RunMode.SIMULATION.value,
             'submit_status': 0   # Submission has succeeded
         }
     )
-    itask.state.status = TASK_STATUS_RUNNING
     return True
 
 

cylc/flow/run_modes/skip.py

@@ -35,6 +35,7 @@
     TASK_OUTPUT_SUBMITTED,
     TASK_OUTPUT_SUCCEEDED,
 )
+from cylc.flow.util import serialise_set
 
 
 if TYPE_CHECKING:
@@ -78,7 +79,7 @@ def submit_task_job(
         itask, {
             'time_submit': now[1],
             'try_num': itask.get_try_num(),
-            'flow_nums': str(list(itask.flow_nums)),
+            'flow_nums': serialise_set(itask.flow_nums),
             'is_manual_submit': itask.is_manual_submit,
             'job_runner_name': RunMode.SKIP.value,
             'platform_name': RunMode.SKIP.value,

cylc/flow/rundb.py

@@ -714,11 +714,12 @@ def select_task_action_timers(self, callback):
         for row_idx, row in enumerate(self.connect().execute(stmt)):
             callback(row_idx, list(row))
 
-    def select_task_job(self, cycle, name, submit_num=None):
+    def select_task_job(
+        self, cycle: str, name: str, submit_num: str | int | None = None
+    ) -> dict[str, Any] | None:
         """Select items from task_jobs by (cycle, name, submit_num).
 
         :return: a dict for mapping keys to the column values
-        :rtype: dict
         """
         keys = []
         for column in self.tables[self.TABLE_TASK_JOBS].columns[3:]:
@@ -737,7 +738,7 @@ def select_task_job(self, cycle, name, submit_num=None):
             '''  # nosec B608
             # * table name is code constant
             # * keys are code constants
-            stmt_args = [cycle, name]
+            stmt_args: list[Any] = [cycle, name]
         else:
             stmt = rf'''
                 SELECT
@@ -752,14 +753,13 @@ def select_task_job(self, cycle, name, submit_num=None):
             # * table name is code constant
             # * keys are code constants
             stmt_args = [cycle, name, submit_num]
-        try:
+        with suppress(sqlite3.DatabaseError):
             for row in self.connect().execute(stmt, stmt_args):
                 ret = {}
                 for key, value in zip(keys, row):
                     ret[key] = value
                 return ret
-        except sqlite3.DatabaseError:
-            return None
+        return None
 
     def select_jobs_for_restart(self, callback):
         """Select from task_pool+task_states+task_jobs for restart.

cylc/flow/scheduler.py

@@ -965,7 +965,7 @@ def process_queued_task_messages(self) -> None:
         """Process incoming task messages for each task proxy.
 
         """
-        messages: 'Dict[str, List[Tuple[Optional[int], TaskMsg]]]' = {}
+        messages: dict[str, list[TaskMsg]] = {}
 
         # Retrieve queued messages
         while self.message_queue.qsize():
@@ -974,28 +974,23 @@ def process_queued_task_messages(self) -> None:
             except Empty:
                 break
             self.message_queue.task_done()
-            tokens = Tokens(task_msg.job_id, relative=True)
             # task ID (job stripped)
-            task_id = tokens.duplicate(job=None).relative_id
-            messages.setdefault(task_id, [])
-            # job may be None (e.g. simulation mode)
-            job = int(tokens['job']) if tokens['job'] else None
-            messages[task_id].append(
-                (job, task_msg)
-            )
+            task_id = task_msg.job_id.duplicate(job=None).relative_id
+            messages.setdefault(task_id, []).append(task_msg)
 
+        unprocessed_messages: List[TaskMsg] = []
         # Poll tasks for which messages caused a backward state change.
-        to_poll_tasks = []
-        for itask in self.pool.get_tasks():
-            message_items = messages.get(itask.identity)
-            if message_items is None:
+        to_poll_tasks: List[TaskProxy] = []
+        for task_id, message_items in messages.items():
+            itask = self.pool._get_task_by_id(task_id)
+            if itask is None:
+                unprocessed_messages.extend(message_items)
                 continue
             should_poll = False
-            del messages[itask.identity]
-            for submit_num, tm in message_items:
+            for tm in message_items:
                 if self.task_events_mgr.process_message(
                     itask, tm.severity, tm.message, tm.event_time,
-                    self.task_events_mgr.FLAG_RECEIVED, submit_num
+                    self.task_events_mgr.FLAG_RECEIVED, tm.job_id.submit_num
                 ):
                     should_poll = True
             if should_poll:
@@ -1006,11 +1001,18 @@ def process_queued_task_messages(self) -> None:
         # Remaining unprocessed messages have no corresponding task proxy.
         # For example, if I manually set a running task to succeeded, the
         # proxy can be removed, but the orphaned job still sends messages.
-        for tms in messages.values():
-            warn = "Undeliverable task messages received and ignored:"
-            for _, msg in tms:
-                warn += f'\n  {msg.job_id}: {msg.severity} - "{msg.message}"'
-            LOG.warning(warn)
+        warn = ""
+        for tm in unprocessed_messages:
+            job_tokens = self.tokens.duplicate(tm.job_id)
+            tdef = self.config.get_taskdef(job_tokens['task'])
+            if not self.task_events_mgr.process_job_message(
+                job_tokens, tdef, tm.message, tm.event_time
+            ):
+                warn += f'\n  {tm.job_id}: {tm.severity} - "{tm.message}"'
+        if warn:
+            LOG.warning(
+                f"Undeliverable task messages received and ignored:{warn}"
+            )
 
     async def process_command_queue(self) -> None:
         """Process queued commands."""