Merge pull request #6963 from cylc/8.5.x-sync

🤖 Merge 8.5.x-sync into master

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

@@ -1350,7 +1350,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

@@ -39,6 +39,7 @@
 from cylc.flow.cycling.integer import IntegerPoint
 from cylc.flow.scheduler import Scheduler
 from cylc.flow.task_state import (
+    TASK_STATUS_PREPARING,
     TASK_STATUS_FAILED,
     TASK_STATUS_RUNNING,
     TASK_STATUS_SUBMITTED,
@@ -758,6 +759,33 @@ async def test_trigger_with_sequential_task(flow, scheduler, run, log_filter):
                 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)
+
+
 async def test_trigger_with_task_selector(flow, scheduler, start, monkeypatch):
     """Test task matching with the trigger command.