Merge pull request #7148 from MetRonnie/trigger-warm-flow-on

Prevent flow-on after triggering pre-startcp tasks after a warm start

Author: oliver-sanders

changes.d/7101.fix.md

@@ -0,0 +1 @@
+Fixed manual group triggering of tasks before the start cycle point not obeying prerequisites within the group.

changes.d/7148.fix.md

@@ -0,0 +1 @@
+Fixed a bug where tasks would flow on from tasks manually triggered before the start cycle point.

cylc/flow/commands.py

@@ -161,6 +161,7 @@ def _remove_matched_tasks(
     schd: 'Scheduler',
     ids: Set[TaskTokens],
     flow_nums: 'FlowNums',
+    warn_unremovable: bool = True,
 ):
     """Remove matched tasks."""
     # Mapping of *relative* task IDs to removed flow numbers:
@@ -267,15 +268,14 @@ def _remove_matched_tasks(
             )
         LOG.info(f"Removed tasks: {', '.join(sorted(tasks_str_list))}")
 
-    if not_removed:
+    if warn_unremovable and not_removed:
         fnums_str = (
             repr_flow_nums(flow_nums, full=True) if flow_nums else ''
         )
         tasks_str = ', '.join(
             sorted(tokens.relative_id for tokens in not_removed)
         )
-        # This often does not indicate an error - e.g. for group trigger.
-        LOG.debug(f"Task(s) not removable: {tasks_str} {fnums_str}")
+        LOG.warning(f"Task(s) not removable: {tasks_str} {fnums_str}")
 
     if removed and schd.pool.compute_runahead():
         schd.pool.release_runahead_tasks()
@@ -806,7 +806,19 @@ def _force_trigger_tasks(
     # Remove all inactive and selected active group members.
     if flow != [FLOW_NONE]:
         # (No need to remove tasks if triggering with no-flow).
-        _remove_matched_tasks(schd, {*active_to_remove, *inactive}, flow_nums)
+        _remove_matched_tasks(
+            schd,
+            ids := {*active_to_remove, *inactive},
+            flow_nums,
+            warn_unremovable=False,
+        )
+        # Record pre-startcp tasks, as these would not normally be spawned -
+        # see https://github.com/cylc/cylc-flow/pull/7148
+        schd.pool.pre_start_tasks_to_trigger.update(
+            (tokens['task'], point)
+            for tokens in ids
+            if (point := get_point(tokens['cycle'])) < schd.config.start_point
+        )
 
         # trigger should override the held state, however, in-group tasks may
         # have previously been held and active in-group tasks will become
@@ -827,7 +839,7 @@ def _force_trigger_tasks(
         icycle = get_point(id_['cycle'])
         in_flow_prereqs = False
         jtask: Optional[TaskProxy] = None
-        if tdef.is_parentless(icycle):
+        if tdef.is_parentless(icycle, cutoff=schd.config.initial_point):
             # Parentless: set all prereqs to spawn the task.
             jtask = schd.pool._set_prereqs_tdef(
                 icycle, tdef,

cylc/flow/cycling/integer.py

@@ -588,6 +588,22 @@ def __lt__(self, other: 'IntegerSequence') -> bool:
                     return True
         return False
 
+    def __repr__(self) -> str:
+        """
+        >>> IntegerSequence('R6/3/P1', '2')
+        <IntegerSequence start=3, stop=8, step=P1, self.p_context_start=2,
+        i_offset=P0>
+        """
+        ret = f"start={self.p_start}, stop={self.p_stop}, step={self.i_step}"
+        if self.p_context_start not in {self.p_start, None}:
+            ret += f", {self.p_context_start=}"
+        if self.p_context_stop not in {self.p_stop, None}:
+            ret += f", {self.p_context_stop=}"
+        for attr in ('i_offset', 'exclusions'):
+            if (value := getattr(self, attr)) is not None:
+                ret += f", {attr}={value}"
+        return f"<{type(self).__name__} {ret}>"
+
 
 def init_from_cfg(_):
     """Placeholder function required by all cycling modules."""

cylc/flow/task_pool.py

@@ -211,8 +211,9 @@ def __init__(
             self.config.runtime['descendants']
         )
 
-        self.tasks_to_hold: Set[Tuple[str, 'PointBase']] = set()
-        self.tasks_to_trigger_now: Set['TaskProxy'] = set()
+        self.tasks_to_hold: set[tuple[str, 'PointBase']] = set()
+        self.tasks_to_trigger_now: set['TaskProxy'] = set()
+        self.pre_start_tasks_to_trigger: set[tuple[str, 'PointBase']] = set()
 
     def set_stop_task(self, task_id):
         """Set stop after a task."""
@@ -837,9 +838,11 @@ def spawn_to_rh_limit(
         longer parentless, and/or hit the runahead limit.
 
         """
-        if not flow_nums or point is None:
-            # Force-triggered no-flow task.
-            # Or called with an invalid next_point.
+        if (
+            not flow_nums  # Force-triggered no-flow task
+            or point is None  # Reached end of sequence?
+            or point < self.config.start_point  # Warm start
+        ):
             return
         if self.runahead_limit_point is None:
             self.compute_runahead()
@@ -848,7 +851,7 @@ def spawn_to_rh_limit(
 
         is_xtrig_sequential = False
         while point is not None and (point <= self.runahead_limit_point):
-            if tdef.is_parentless(point):
+            if tdef.is_parentless(point, cutoff=self.config.start_point):
                 ntask, is_in_pool, is_xtrig_sequential = (
                     self.get_or_spawn_task(point, tdef, flow_nums)
                 )
@@ -866,7 +869,11 @@ def spawn_to_rh_limit(
 
     def spawn_if_parentless(self, tdef, point, flow_nums):
         """Spawn a task if parentless, regardless of runahead limit."""
-        if flow_nums and point is not None and tdef.is_parentless(point):
+        if (
+            flow_nums
+            and point is not None
+            and tdef.is_parentless(point, cutoff=self.config.start_point)
+        ):
             ntask, is_in_pool, _ = self.get_or_spawn_task(
                 point, tdef, flow_nums
             )
@@ -906,6 +913,9 @@ def remove(self, itask: 'TaskProxy', reason: Optional[str] = None) -> None:
             pass
         else:
             self.tasks_to_trigger_now.discard(itask)
+            self.pre_start_tasks_to_trigger.discard(
+                (itask.tdef.name, itask.point)
+            )
             self.tasks_removed = True
             self.active_tasks_changed = True
             if not self.active_tasks[itask.point]:
@@ -1727,8 +1737,8 @@ def can_be_spawned(self, name: str, point: 'PointBase') -> bool:
         return True
 
     def _get_task_history(
-        self, name: str, point: 'PointBase', flow_nums: Set[int]
-    ) -> Tuple[int, Optional[str], bool]:
+        self, name: str, point: 'PointBase', flow_nums: 'FlowNums'
+    ) -> tuple[int, str | None, bool]:
         """Get submit_num, status, flow_wait for point/name in flow_nums.
 
         Args:
@@ -1766,7 +1776,10 @@ def _get_task_history(
         return submit_num, status, flow_wait
 
     def _load_historical_outputs(self, itask: 'TaskProxy') -> None:
-        """Load a task's historical outputs from the DB."""
+        """Load a task's historical outputs from the DB.
+
+        NOTE this creates a task_states/task_outputs DB entry if not present.
+        """
         info = self.workflow_db_mgr.pri_dao.select_task_outputs(
             itask.tdef.name, str(itask.point))
         if not info:
@@ -1803,9 +1816,9 @@ def spawn_task(
         self,
         name: str,
         point: 'PointBase',
-        flow_nums: Set[int],
+        flow_nums: 'FlowNums',
         flow_wait: bool = False,
-    ) -> Optional[TaskProxy]:
+    ) -> TaskProxy | None:
         """Return a new task proxy for the given flow if possible.
 
         We need to hit the DB for:
@@ -1826,8 +1839,18 @@ def spawn_task(
             self._get_task_history(name, point, flow_nums)
         )
 
+        if (
+            not prev_status
+            and point < self.config.start_point
+            and flow_nums.issuperset({1})
+            # Warm start - treat pre-startcp tasks as already run in flow=1,
+            # unless manually triggered:
+            and (name, point) not in self.pre_start_tasks_to_trigger
+        ):
+            return None
+
         # Create the task proxy with any completed outputs loaded.
-        itask = self._get_task_proxy_db_outputs(
+        itask = self._load_db_task_proxy(
             point,
             self.config.get_taskdef(name),
             flow_nums,
@@ -1925,7 +1948,7 @@ def _spawn_after_flow_wait(self, itask: TaskProxy) -> None:
         self.workflow_db_mgr.put_update_task_flow_wait(itask)
         return None
 
-    def _get_task_proxy_db_outputs(
+    def _load_db_task_proxy(
         self,
         point: 'PointBase',
         taskdef: 'TaskDef',
@@ -1935,8 +1958,11 @@ def _get_task_proxy_db_outputs(
         transient: bool = False,
         is_manual_submit: bool = False,
         submit_num: int = 0,
-    ) -> Optional['TaskProxy']:
-        """Spawn a task, update outputs from DB."""
+    ) -> 'TaskProxy | None':
+        """Spawn a task, update outputs from DB.
+
+        NOTE this creates a task_states/task_outputs DB entry if not present.
+        """
 
         if not self.can_be_spawned(taskdef.name, point):
             return None
@@ -2140,7 +2166,7 @@ def set_prereqs_and_outputs(
                     no_op = False
                 else:
                     # Outputs (may be empty list)
-                    trans = self._get_task_proxy_db_outputs(
+                    trans = self._load_db_task_proxy(
                         icycle, tdef, flow_nums,
                         flow_wait=flow_wait, transient=True
                     )

cylc/flow/task_proxy.py

@@ -218,8 +218,8 @@ def __init__(
         self,
         scheduler_tokens: 'Tokens',
         tdef: 'TaskDef',
-        start_point: 'PointBase',
-        flow_nums: Optional['FlowNums'] = None,
+        point: 'PointBase',
+        flow_nums: 'FlowNums | None' = None,
         status: str = TASK_STATUS_WAITING,
         is_held: bool = False,
         submit_num: int | None = 0,
@@ -242,7 +242,7 @@ def __init__(
             # (don't share flow_nums ref with parent task)
             self.flow_nums = flow_nums.copy()
         self.flow_wait = flow_wait
-        self.point = start_point
+        self.point = point
         self.tokens: TaskTokens = scheduler_tokens.duplicate(
             cycle=str(self.point),
             task=self.tdef.name,
@@ -294,7 +294,9 @@ def __init__(
         # Set xtrigger checking type, which effects parentless spawning.
         self.is_xtrigger_sequential = bool(
             sequential_xtrigger_labels
-            and self.tdef.is_parentless(start_point)
+            and self.tdef.is_parentless(
+                self.point, cutoff=self.tdef.initial_point
+            )
             and sequential_xtrigger_labels.intersection(self.state.xtriggers)
         )
 

cylc/flow/task_trigger.py

@@ -239,6 +239,11 @@ def get_prerequisite(
 
         # Loop over TaskTrigger instances.
         for task_trigger in self.task_triggers:
+            key = (
+                task_trigger.get_point(point),
+                task_trigger.task_name,
+                task_trigger.output,
+            )
             if task_trigger.cycle_point_offset is not None:
                 # Compute trigger cycle point from offset.
                 if task_trigger.offset_is_from_icp:
@@ -247,30 +252,26 @@ def get_prerequisite(
                 else:
                     prereq_offset_point = get_point_relative(
                         task_trigger.cycle_point_offset, point)
+                if prereq_offset_point < tdef.initial_point:
+                    # Pre-initial dependency - treat as satisfied
+                    cpre[key] = True
+                    continue
                 if prereq_offset_point > point:
                     # Update tdef.max_future_prereq_offset.
                     prereq_offset = prereq_offset_point - point
-                    if (tdef.max_future_prereq_offset is None or
-                            (prereq_offset >
-                             tdef.max_future_prereq_offset)):
-                        tdef.max_future_prereq_offset = (
-                            prereq_offset)
-                cpre[(
-                    task_trigger.get_point(point),
-                    task_trigger.task_name,
-                    task_trigger.output
-                )] = (
-                    (prereq_offset_point < tdef.start_point) &
-                    (point >= tdef.start_point)
+                    if (
+                        tdef.max_future_prereq_offset is None
+                        or prereq_offset > tdef.max_future_prereq_offset
+                    ):
+                        tdef.max_future_prereq_offset = prereq_offset
+                cpre[key] = (
+                    prereq_offset_point < tdef.start_point
+                    and point >= tdef.start_point
                 )
             else:
                 # Trigger is within the same cycle point.
                 # Register task message with Prerequisite object.
-                cpre[(
-                    task_trigger.get_point(point),
-                    task_trigger.task_name,
-                    task_trigger.output,
-                )] = False
+                cpre[key] = False
         cpre.set_conditional_expr(self.get_expression(point))
         return cpre
 
@@ -299,6 +300,14 @@ def __str__(self):
                 ret.append('( %s )' % str(item))
         return ' '.join(ret)
 
+    def __repr__(self) -> str:
+        """
+        >>> from unittest.mock import Mock
+        >>> Dependency(exp=[Mock()], task_triggers=[Mock()], suicide=False)
+        <Dependency ...>
+        """
+        return f"<{type(self).__name__} {self}>"
+
     @classmethod
     def _stringify_list(cls, nested_expr, point):
         """Stringify a nested list of TaskTrigger objects."""

cylc/flow/taskdef.py

@@ -419,18 +419,24 @@ def next_point(self, point):
             p_next = min(adjusted)
         return p_next
 
-    def is_parentless(self, point):
-        """Return True if task has no parents at point.
+    def is_parentless(self, point: 'PointBase', cutoff: 'PointBase') -> bool:
+        """Return True if task has no parents at the given point.
 
         Tasks are considered parentless if they have:
           - no parents at all
-          - all parents < initial cycle point
+          - all parents < cutoff cycle point
           - only absolute triggers
 
         Absolute-triggered tasks are auto-spawned like true parentless tasks,
         (once the trigger is satisfied they are effectively parentless) but
         with a prerequisite that gets satisfied when the absolute output is
         completed at runtime.
+
+        Args:
+            point: The cycle point to check.
+            cutoff: This should be the start cycle point for the startup
+                spawning, or the intial cycle point for manually triggered
+                tasks.
         """
         if not self.graph_parents:
             # No parents at any point
@@ -441,7 +447,7 @@ def is_parentless(self, point):
         parent_points = self.get_parent_points(point)
         return (
             not parent_points
-            or all(x < self.start_point for x in parent_points)
+            or all(x < cutoff for x in parent_points)
             or self.has_only_abs_triggers(point)
         )