Flow trigger CLI and defaults

CHANGES.md

@@ -95,6 +95,13 @@ line when specifying a Cylc ID that includes your username (e.g. `'~user/workflo
 [#4737](https://github.com/cylc/cylc-flow/pull/4737) -
 Fix issue which prevented tasks with incomplete outputs from being rerun by
 subsequent flows.
+### Enhancements
+
+[#4738](https://github.com/cylc/cylc-flow/pull/4738) and
+[#4739](https://github.com/cylc/cylc-flow/pull/4739) - Implement `cylc trigger
+[--flow=] [--wait]` for manual triggering with respect to active flows (the
+default), specific flows, new flows, or one-off task runs.
+
 
 -------------------------------------------------------------------------------
 ## __cylc-8.0rc1 (<span actions:bind='release-date'>Released 2022-02-17</span>)__

cylc/flow/data_messages.proto

@@ -218,6 +218,7 @@ message PbTaskProxy {
     map<string, PbTrigger> xtriggers = 24;
     optional bool is_queued = 25;
     optional bool is_runahead = 26;
+    optional bool flow_wait = 27;
 }
 
 message PbFamily {

cylc/flow/data_store_mgr.py

@@ -1080,7 +1080,7 @@ def apply_task_proxy_db_history(self):
                         TASK_STATUS_SUCCEEDED
                     )
             ):
-                for message in json.loads(outputs_str).values():
+                for message in json.loads(outputs_str):
                     itask.state.outputs.set_completion(message, True)
             # Gather tasks with flow id.
             prereq_ids.add(f'{tokens.relative_id}/{flow_nums_str}')

cylc/flow/dbstatecheck.py

@@ -153,7 +153,7 @@ def task_state_met(self, task, cycle, status=None, message=None):
             return any(
                 message == value
                 for outputs_str, in res
-                for value in json.loads(outputs_str).values()
+                for value in json.loads(outputs_str)
             )
 
     @staticmethod

cylc/flow/flow_mgr.py

@@ -24,6 +24,10 @@
 
 
 FlowNums = Set[int]
+# Flow constants
+FLOW_ALL = "all"
+FLOW_NEW = "new"
+FLOW_NONE = "none"
 
 
 class FlowMgr:

cylc/flow/log_diagnosis.py

@@ -28,17 +28,17 @@
     from cylc.flow.scheduler import Scheduler
 
 
-RE_TRIG = re.compile(r'(.*? -triggered off \[.*\])$')
+RE_TRIG = re.compile(r'(^.*? -triggered off \[.*\].*$)')
 
 
 def run_reftest(schd: 'Scheduler') -> None:
     """Run reference test at shutdown."""
     reffilename = schd.config.get_ref_log_name()
     curfilename = get_workflow_test_log_name(schd.workflow)
-    ref = _load_reflog(reffilename)
-    cur = _load_reflog(curfilename)
+    ref = _load_reflog(reffilename, False)
     if not ref:
         raise WorkflowEventError("No triggering events in reference log.")
+    cur = _load_reflog(curfilename, "in flow" not in ref[0])
     if not cur:
         raise WorkflowEventError("No triggering events in test log.")
     if ref == cur:
@@ -53,13 +53,19 @@ def run_reftest(schd: 'Scheduler') -> None:
         raise exc
 
 
-def _load_reflog(filename):
-    """Reference test: get trigger info from reference log."""
+def _load_reflog(filename, strip_flows):
+    """Reference test: get trigger info from reference log.
+
+    Back-compat for old logs: strip flow nums from each line.
+    """
     res = []
     with open(os.path.expandvars(filename), 'r') as reflog:
         for line in reflog:
             match = RE_TRIG.search(line)
             if match:
-                res.append(match.groups()[0])
+                if strip_flows:
+                    res.append(re.sub(' in flow .*$', '', match.groups()[0]))
+                else:
+                    res.append(match.groups()[0])
     res.sort()
     return res

cylc/flow/network/resolvers.py

@@ -813,23 +813,25 @@ def stop(
 
     def force_trigger_tasks(
         self,
-        tasks=None,
-        reflow=False,
-        flow_descr=None,
+        tasks: Iterable[str],
+        flow: Iterable[str],
+        flow_wait: bool,
+        flow_descr: Optional[str] = None,
     ):
         """Trigger submission of task jobs where possible.
 
         Args:
             tasks (list):
                 List of identifiers or task globs.
-            reflow (bool):
-                Start new flow from triggered tasks.
+            flow (list):
+                Flow ownership of triggered tasks.
+            flow_wait (bool):
+                Wait for flows before continuing
             flow_descr (str):
                 Description of new flow.
 
         Returns:
             tuple: (outcome, message)
-
             outcome (bool)
                 True if command successfully queued.
             message (str)
@@ -841,7 +843,8 @@ def force_trigger_tasks(
                 "force_trigger_tasks",
                 (tasks or [],),
                 {
-                    "reflow": reflow,
+                    "flow": flow,
+                    "flow_wait": flow_wait,
                     "flow_descr": flow_descr
                 }
             ),

cylc/flow/network/schema.py

@@ -39,6 +39,7 @@
 from graphene.utils.str_converters import to_snake_case
 
 from cylc.flow.broadcast_mgr import ALL_CYCLE_POINTS_STRS, addict
+from cylc.flow.flow_mgr import FLOW_ALL, FLOW_NEW, FLOW_NONE
 from cylc.flow.id import Tokens
 from cylc.flow.task_outputs import SORT_ORDERS
 from cylc.flow.task_state import (
@@ -865,6 +866,7 @@ class Meta:
     is_queued = Boolean()
     is_runahead = Boolean()
     flow_nums = String()
+    flow_wait = Boolean()
     depth = Int()
     job_submits = Int()
     outputs = graphene.List(
@@ -1404,6 +1406,11 @@ def description(self):
         return StopMode(self.value).describe()
 
 
+class Flow(String):
+    """An integer or one of {FLOW_ALL}, {FLOW_NEW} or {FLOW_NONE}."""
+    # (Note docstrings can't be f-strings).
+
+
 # Mutations:
 
 # TODO: re-instate:
@@ -1742,6 +1749,27 @@ class Arguments:
     result = GenericScalar()
 
 
+class FlowMutationArguments:
+    flow = graphene.List(
+        graphene.NonNull(Flow),
+        default_value=[FLOW_ALL],
+        description=sstrip(f'''
+            The flow(s) to trigger these tasks in.
+
+            This should be a list of flow numbers OR a single-item list
+            containing one of the following three strings:
+
+            Alternatively this may be a single-item list containing one of
+            the following values:
+
+            * {FLOW_ALL} - Triggered tasks belong to all active flows
+              (default).
+            * {FLOW_NEW} - Triggered tasks are assigned to a new flow.
+            * {FLOW_NONE} - Triggered tasks do not belong to any flow.
+        ''')
+    )
+
+
 class Hold(Mutation, TaskMutation):
     class Meta:
         description = sstrip('''
@@ -1832,9 +1860,30 @@ class Meta:
         ''')
         resolver = partial(mutator, command='force_trigger_tasks')
 
-    class Arguments(TaskMutation.Arguments):
-        reflow = Boolean()
-        flow_descr = String()
+    class Arguments(TaskMutation.Arguments, FlowMutationArguments):
+        flow_wait = Boolean(
+            default_value=False,
+            description=sstrip('''
+                Should the workflow "wait" or "continue on" from this task?
+
+                If `false` the scheduler will spawn and run downstream tasks
+                as normal (default).
+
+                If `true` the scheduler will not spawn the downstream tasks
+                unless it has been caught by the same flow at a later time.
+
+                For example you might set this to True to trigger a task
+                ahead of a flow, where you don't want the scheduled to
+                "continue on" from this task until the flow has caught up
+                with it.
+            ''')
+        )
+        flow_descr = String(
+            description=sstrip('''
+                If starting a new flow, this field can be used to provide the
+                new flow with a description for later reference.
+            ''')
+        )
 
 
 def _mut_field(cls):

cylc/flow/rundb.py

@@ -23,6 +23,7 @@
 from typing import List, Tuple
 
 from cylc.flow import LOG
+from cylc.flow.util import deserialise
 import cylc.flow.flags
 
 
@@ -263,6 +264,7 @@ class CylcWorkflowDAO:
         TABLE_TASK_OUTPUTS: [
             ["cycle", {"is_primary_key": True}],
             ["name", {"is_primary_key": True}],
+            ["flow_nums", {"is_primary_key": True}],
             ["outputs"],
         ],
         TABLE_TASK_POOL: [
@@ -293,6 +295,7 @@ class CylcWorkflowDAO:
             ["time_updated"],
             ["submit_num", {"datatype": "INTEGER"}],
             ["status"],
+            ["flow_wait", {"datatype": "INTEGER"}],
         ],
         TABLE_TASK_TIMEOUT_TIMERS: [
             ["cycle", {"is_primary_key": True}],
@@ -709,11 +712,7 @@ def select_submit_nums(self, name, point):
 
         Fetch submit number and flow_nums for spawning tasks.
 
-        Return:
-            {
-                flow_nums: submit_num,
-                ...,
-            }
+        Return: {submit_num: (flow_wait, flow_nums)}
 
         Args:
             name: task name
@@ -724,13 +723,32 @@ def select_submit_nums(self, name, point):
         # Not an injection, simply putting the table name in the SQL query
         # expression as a string constant local to this module.
         stmt = (  # nosec
-            r"SELECT flow_nums,submit_num FROM %(name)s"
+            r"SELECT flow_nums,submit_num,flow_wait FROM %(name)s"
             r" WHERE name==? AND cycle==?"
         ) % {"name": self.TABLE_TASK_STATES}
         ret = {}
-        for flow_nums, submit_num in self.connect().execute(
+        for flow_nums_str, submit_num, flow_wait in self.connect().execute(
                 stmt, (name, point,)):
-            ret[flow_nums] = submit_num
+            ret[submit_num] = (flow_wait == 1, deserialise(flow_nums_str))
+        return ret
+
+    def select_task_outputs(self, name, point):
+        """Select task outputs for each flow.
+
+        Return: {outputs_list: flow_nums_set}
+
+        """
+        stmt = rf'''
+            SELECT
+               flow_nums,outputs
+            FROM
+               {self.TABLE_TASK_OUTPUTS}
+            WHERE
+                name==? AND cycle==?
+        '''  # nosec (table name is code constant)
+        ret = {}
+        for flow_nums, outputs in self.connect().execute(stmt, (name, point,)):
+            ret[outputs] = deserialise(flow_nums)
         return ret
 
     def select_xtriggers_for_restart(self, callback):

cylc/flow/scheduler.py

@@ -50,7 +50,7 @@
 from cylc.flow.cycling.loader import get_point
 from cylc.flow.data_store_mgr import DataStoreMgr
 from cylc.flow.id import Tokens
-from cylc.flow.flow_mgr import FlowMgr
+from cylc.flow.flow_mgr import FLOW_NONE, FlowMgr, FLOW_NEW
 from cylc.flow.exceptions import (
     CommandFailedError,
     CyclingError,
@@ -681,7 +681,7 @@ def _load_pool_from_tasks(self):
         # flow number set in this call:
         self.pool.force_trigger_tasks(
             self.options.starttask,
-            reflow=True,
+            flow=[FLOW_NEW],
             flow_descr=f"original flow from {self.options.starttask}"
         )
 
@@ -1300,10 +1300,14 @@ def release_queued_tasks(self):
                 self.client_pub_key_dir,
                 self.config.run_mode('simulation')
             ):
-                # (Not using f"{itask}"_here to avoid breaking func tests)
+                if itask.flow_nums:
+                    flow = ','.join(str(i) for i in itask.flow_nums)
+                else:
+                    flow = FLOW_NONE
                 meth(
                     f"{itask.identity} -triggered off "
                     f"{itask.state.get_resolved_dependencies()}"
+                    f" in flow {flow}"
                 )
 
     def process_workflow_db_queue(self):
@@ -1884,9 +1888,10 @@ def resume_workflow(self, quiet: bool = False) -> None:
         self.workflow_db_mgr.delete_workflow_paused()
         self.update_data_store()
 
-    def command_force_trigger_tasks(self, items, reflow, flow_descr):
-        """Trigger tasks."""
-        return self.pool.force_trigger_tasks(items, reflow, flow_descr)
+    def command_force_trigger_tasks(self, items, flow, flow_wait, flow_descr):
+        """Manual task trigger."""
+        return self.pool.force_trigger_tasks(
+            items, flow, flow_wait, flow_descr)
 
     def command_force_spawn_children(self, items, outputs, flow_num):
         """Force spawn task successors.

cylc/flow/scripts/cylc.py

@@ -212,8 +212,8 @@ def get_version(long=False):
         'use standard tools to inspect the environment'
         ' e.g. https://pypi.org/project/pipdeptree/',
     'checkpoint':
-        'DB checkpoints have been removed, use a reflow to '
-        '"rewind" a workflow.',
+        'DB checkpoints have been removed. You can now "rewind" a'
+        ' workflow by triggering the flow anywhere in the graph.',
     'conditions':
         'cylc conditions has been replaced by cylc help license',
     'documentation':

cylc/flow/scripts/set_outputs.py

@@ -24,8 +24,7 @@
 This allows you to manually intervene with Cylc's scheduling algorithm by
 artificially satisfying outputs of tasks.
 
-If a flow number is given, the child tasks will start (or continue) that flow,
-otherwise no reflow will occur.
+If a flow number is given, the child tasks will start (or continue) that flow.
 
 Examples:
   # For example, for the following dependency graph: