Always submit preparing tasks. (#6768)

Remote platform init was thwarting the trigger-if-paused logic.

Author: hjoliver

changes.d/6768.fix.md

@@ -0,0 +1 @@
+Fix a bug causing remote tasks to stay in the preparing state if triggered when the workflow is paused.

cylc/flow/scheduler.py

@@ -1354,7 +1354,13 @@ def release_tasks_to_run(self) -> bool:
                 pre_prep_tasks.update(self.pool.tasks_to_trigger_now)
                 self.pool.tasks_to_trigger_now = set()
 
-            if not self.is_paused:
+            if self.is_paused:
+                # finish processing preparing tasks
+                pre_prep_tasks.update({
+                    itask for itask in self.pool.get_tasks()
+                    if itask.state(TASK_STATUS_PREPARING)
+                })
+            else:
                 # release queued tasks
                 pre_prep_tasks.update(self.pool.release_queued_tasks())
                 if self.pool.tasks_to_trigger_on_resume:

tests/integration/test_force_trigger.py

@@ -38,6 +38,7 @@
 )
 from cylc.flow.cycling.integer import IntegerPoint
 from cylc.flow.task_state import (
+    TASK_STATUS_PREPARING,
     TASK_STATUS_WAITING,
     TASK_STATUS_RUNNING
 )
@@ -752,3 +753,30 @@ async def test_trigger_with_sequential_task(flow, scheduler, run, log_filter):
                 if log_filter(contains='[2/foo/02:running] (received)failed'):
                     break
                 await asyncio.sleep(0)
+
+
+async def test_trigger_whilst_paused_preparing(flow, scheduler, run, complete):
+    """It should run "preparing" tasks even if the workflow is paused.
+
+    Remote init leaves tasks as preparing for a while. These must still be
+    pushed through to running, even if triggered while the workflow is paused.
+
+    See https://github.com/cylc/cylc-flow/pull/6768
+
+    """
+    id_ = flow(
+        {
+            'scheduling': {
+                'graph': {'R1': 'a'},
+            },
+        }
+    )
+    schd = scheduler(id_)
+    async with run(schd):
+
+        # Instead of triggering, artificially set it as preparing.
+        a = schd.pool.get_tasks()[0]
+        a.state_reset(TASK_STATUS_PREPARING)
+
+        # 1/a should run even though the workflow is paused.
+        await complete(schd, '1/a', allow_paused=True, timeout=1)