Preparing vs active

cylc/flow/scheduler.py

@@ -737,6 +737,8 @@ def restart_remote_init(self):
             itask.platform['install target'] = (
                 get_install_target_from_platform(itask.platform))
             if (
+                # we don't need to remote-init for preparing tasks because
+                # they will be reset to waiting on restart
                 itask.state(*TASK_STATUSES_ACTIVE)
                 and not (
                     is_platform_with_target_in_list(

cylc/flow/task_pool.py

@@ -822,7 +822,7 @@ def reload_taskdefs(self) -> None:
                 itask.copy_to_reload_successor(new_task)
                 self._swap_out(new_task)
                 LOG.info(f"[{itask}] reloaded task definition")
-                if itask.state(*TASK_STATUSES_ACTIVE):
+                if itask.state(*TASK_STATUSES_ACTIVE, TASK_STATUS_PREPARING):
                     LOG.warning(
                         f"[{itask}] active with pre-reload settings"
                     )
@@ -898,6 +898,8 @@ def can_stop(self, stop_mode):
                 and itask.state(*TASK_STATUSES_ACTIVE)
                 and not itask.state.kill_failed
             )
+            # we don't need to check for preparing tasks because they will be
+            # reset to waiting on restart
             for itask in self.get_tasks()
         )
 

tests/integration/test_task_pool.py

@@ -438,3 +438,34 @@ async def test_trigger_states(status, should_trigger, one, run):
 
         # check whether the task triggered
         assert task.is_manual_submit == should_trigger
+
+
+async def test_preparing_tasks_on_restart(one_conf, flow, scheduler, run):
+    """Preparing tasks should be reset to waiting on restart.
+
+    This forces preparation to be re-done on restart so that it uses the
+    new configuration.
+
+    See discussion on: https://github.com/cylc/cylc-flow/pull/4668
+
+    """
+    id_ = flow(one_conf)
+
+    # start the workflow, reset a task to preparing
+    one = scheduler(id_)
+    async with run(one):
+        task = one.pool.filter_task_proxies(['*'])[0][0]
+        task.state.reset(TASK_STATUS_PREPARING)
+
+    # when we restart the task should have been reset to waiting
+    one = scheduler(id_)
+    async with run(one):
+        task = one.pool.filter_task_proxies(['*'])[0][0]
+        assert task.state(TASK_STATUS_WAITING)
+        task.state.reset(TASK_STATUS_SUCCEEDED)
+
+    # whereas if we reset the task to succeeded the state is not reset
+    one = scheduler(id_)
+    async with run(one):
+        task = one.pool.filter_task_proxies(['*'])[0][0]
+        assert task.state(TASK_STATUS_SUCCEEDED)