Merge pull request #6940 from MetRonnie/crash

Fix scheduler crash after broadcasting `run mode = skip`

Author: MetRonnie

changes.d/6940.fix.md

@@ -0,0 +1 @@
+Fixed a scheduler crash that could occur after re-running a task in skip mode.

cylc/flow/task_events_mgr.py

@@ -548,7 +548,7 @@ def _get_remote_conf(self, itask, key):
             itask.platform[key]
         )
 
-    def _get_workflow_platforms_conf(self, itask, key):
+    def _get_workflow_platforms_conf(self, itask: 'TaskProxy', key: str):
         """Return top level [runtime] items that default to platforms."""
         overrides = self.broadcast_mgr.get_broadcast(itask.tokens)
         return (
@@ -1844,7 +1844,7 @@ def _get_handler_template_variables(
         }
         # fmt: on
 
-    def _reset_job_timers(self, itask):
+    def _reset_job_timers(self, itask: 'TaskProxy'):
         """Set up poll timer and timeout for task."""
 
         if itask.transient:
@@ -1898,18 +1898,18 @@ def _reset_job_timers(self, itask):
         itask.poll_timer = TaskActionTimer(ctx=ctx, delays=delays)
         # Log timeout and polling schedule
         message = f"health: {timeout_key}={timeout_str}"
-        # Attempt to group identical consecutive delays as N*DELAY,...
         if itask.poll_timer.delays:
-            items = []  # [(number of item - 1, item), ...]
+            # Group identical consecutive delays as N*DELAY,...
+            items: List[List[float]] = []  # [[number of item, item], ...]
             for delay in itask.poll_timer.delays:
                 if items and items[-1][1] == delay:
                     items[-1][0] += 1
                 else:
-                    items.append([0, delay])
+                    items.append([1, delay])
             message += ', polling intervals='
             for num, item in items:
-                if num:
-                    message += '%d*' % (num + 1)
+                if num > 1:
+                    message += f'{num}*'
                 message += '%s,' % intvl_as_str(item)
             message += '...'
         LOG.debug(f"[{itask}] {message}")
@@ -1920,7 +1920,7 @@ def _reset_job_timers(self, itask):
     def process_execution_polling_intervals(
         polling_intervals: List[float],
         time_limit: float,
-        time_limit_polling_intervals: List[float]
+        time_limit_polling_intervals: Optional[List[float]]
     ) -> List[float]:
         """Create a list of polling times.
 
@@ -1952,6 +1952,11 @@ def process_execution_polling_intervals(
         >>> this([], 10, [5])
         [15, 5]
 
+        # There are no execution time limit polling intervals set - just
+        # repeat the execution polling interval until the time limit:
+        >>> this([10], 25, None)
+        [10, 10]
+
         # We have a list of execution time limit polling intervals,
         >>> this([10], 25, [5, 6, 7, 8])
         [10, 10, 10, 6, 7, 8]
@@ -1968,17 +1973,19 @@ def process_execution_polling_intervals(
             size = int((time_limit - sum(delays)) / delays[-1])
             delays.extend([delays[-1]] * size)
 
-        # After the last delay before the execution time limit add the
-        # delay to get to the execution_time_limit
-        if len(time_limit_polling_intervals) == 1:
-            time_limit_polling_intervals.append(
-                time_limit_polling_intervals[0]
-            )
-        time_limit_polling_intervals[0] += time_limit - sum(delays)
+        if time_limit_polling_intervals:
+            # After the last delay before the execution time limit add the
+            # delay to get to the execution_time_limit
+            if len(time_limit_polling_intervals) == 1:
+                time_limit_polling_intervals.append(
+                    time_limit_polling_intervals[0]
+                )
+            time_limit_polling_intervals[0] += time_limit - sum(delays)
+
+            # After the execution time limit, poll at the
+            # execution time limit polling intervals.
+            delays += time_limit_polling_intervals
 
-        # After the execution time limit poll at execution time limit polling
-        # intervals.
-        delays += time_limit_polling_intervals
         return delays
 
     def add_event_timer(self, id_key: EventKey, event_timer) -> None:

tests/integration/run_modes/test_skip.py

@@ -17,6 +17,12 @@
 """
 
 from cylc.flow.cycling.integer import IntegerPoint
+from cylc.flow.scheduler import Scheduler
+from cylc.flow.task_outputs import TASK_OUTPUT_FAILED
+from cylc.flow.task_state import (
+    TASK_STATUS_FAILED,
+    TASK_STATUS_SUCCEEDED,
+)
 
 
 async def test_settings_override_from_broadcast(
@@ -237,3 +243,34 @@ async def test_outputs_can_be_changed(
             ['1'], ['one'], [{'skip': {'outputs': 'succeeded'}}]
         )
         schd.submit_task_jobs(schd.pool.get_tasks())
+
+
+async def test_rerun_after_skip_mode_broadcast(
+    flow, one_conf, scheduler, start
+):
+    """Test re-running a task after it has been set to skip.
+
+    See https://github.com/cylc/cylc-flow/pull/6940
+    """
+    id_ = flow({
+        **one_conf,
+        "runtime": {
+            "one": {
+                "execution time limit": "PT1M",
+            },
+        },
+    })
+    schd: Scheduler = scheduler(id_, run_mode='live')
+    async with start(schd):
+        itask = schd.pool.get_tasks()[0]
+        schd.submit_task_jobs([itask])
+        schd.task_events_mgr.process_message(
+            itask, 'CRITICAL', TASK_OUTPUT_FAILED
+        )
+        assert itask.state(TASK_STATUS_FAILED)
+
+        schd.broadcast_mgr.put_broadcast(
+            ['1'], ['root'], [{'run mode': 'skip'}]
+        )
+        schd.submit_task_jobs([itask])
+        assert itask.state(TASK_STATUS_SUCCEEDED)