Merge 'upstream/master' into cylc-remove

.github/workflows/build.yml

@@ -18,8 +18,10 @@ jobs:
     strategy:
       matrix:
         os: ['ubuntu-latest']
-        python: ['3.7', '3.8', '3.9', '3.10', '3.11']
+        python: ['3.8', '3.9', '3.10', '3.11']
         include:
+          - os: 'ubuntu-22.04'
+            python: '3.7'
           - os: 'macos-latest'
             python: '3.8'
     steps:

.github/workflows/test_fast.yml

@@ -20,9 +20,10 @@ jobs:
       fail-fast: false # don't stop on first failure
       matrix:
         os: ['ubuntu-latest']
-        python-version: ['3.7', '3.8', '3.10', '3.11', '3']
+        python-version: ['3.8', '3.10', '3.11', '3']
         include:
-          # mac os test
+          - os: 'ubuntu-22.04'
+            python-version: '3.7'
           - os: 'macos-latest'
             python-version: '3.9'  # oldest supported version
           # non-utc timezone test

.github/workflows/test_functional.yml

@@ -40,7 +40,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: ['ubuntu-latest']
+        os: ['ubuntu-22.04']
         python-version: ['3.7']
         test-base: ['tests/f']
         chunk: ['1/4', '2/4', '3/4', '4/4']
@@ -56,20 +56,20 @@ jobs:
             platform: '_local_background*'
           # tests/k
           - name: 'flaky'
-            os: 'ubuntu-latest'
+            os: 'ubuntu-22.04'
             python-version: '3.7'
             test-base: 'tests/k'
             chunk: '1/1'
             platform: '_local_background* _local_at*'
           # remote platforms
           - name: '_remote_background_indep_poll'
-            os: 'ubuntu-latest'
+            os: 'ubuntu-22.04'
             python-version: '3.7'
             test-base: 'tests/f tests/k'
             chunk: '1/1'
             platform: '_remote_background_indep_poll _remote_at_indep_poll'
           - name: '_remote_background_indep_tcp'
-            os: 'ubuntu-latest'
+            os: 'ubuntu-22.04'
             test-base: 'tests/f tests/k'
             python-version: '3.7'
             chunk: '1/1'

.github/workflows/test_tutorial_workflow.yml

@@ -21,8 +21,12 @@ jobs:
   test:
     strategy:
       matrix:
-        python-version: ['3.7', '3']
-    runs-on: ubuntu-latest
+        include:
+          - os: 'ubuntu-latest'
+            python-version: '3'
+          - os: 'ubuntu-22.04'
+            python-version: '3.7'
+    runs-on: ${{ matrix.os }}
     timeout-minutes: 10
     steps:
       - name: configure python

changes.d/6316.fix.md

@@ -0,0 +1 @@
+Fixed bug in `cylc vr` where an initial cycle point of `now`/`next()`/`previous()` would result in an error.

changes.d/6367.fix.md

@@ -0,0 +1 @@
+Fix bug where `cylc trigger` and `cylc set` would assign active flows to existing tasks by default.

changes.d/6397.fix.md

@@ -0,0 +1 @@
+Fix "dictionary changed size during iteration error" which could occur with broadcasts.

cylc/flow/command_validation.py

@@ -41,7 +41,7 @@
     f"or '{FLOW_NONE}'"
 )
 ERR_OPT_FLOW_VAL_2 = f"Flow values must be an integer, or '{FLOW_ALL}'"
-ERR_OPT_FLOW_INT = "Multiple flow options must all be integer valued"
+ERR_OPT_FLOW_COMBINE = "Cannot combine --flow={0} with other flow values"
 ERR_OPT_FLOW_WAIT = (
     f"--wait is not compatible with --flow={FLOW_NEW} or --flow={FLOW_NONE}"
 )
@@ -54,44 +54,52 @@ def flow_opts(
 ) -> None:
     """Check validity of flow-related CLI options.
 
-    Note the schema defaults flows to ["all"].
+    Note the schema defaults flows to [].
 
     Examples:
         Good:
+        >>> flow_opts([], False)
         >>> flow_opts(["new"], False)
         >>> flow_opts(["1", "2"], False)
         >>> flow_opts(["1", "2"], True)
 
         Bad:
         >>> flow_opts(["none", "1"], False)
         Traceback (most recent call last):
-        cylc.flow.exceptions.InputError: ... must all be integer valued
+        cylc.flow.exceptions.InputError: Cannot combine --flow=none with other
+        flow values
 
         >>> flow_opts(["cheese", "2"], True)
         Traceback (most recent call last):
         cylc.flow.exceptions.InputError: ... or 'all', 'new', or 'none'
 
         >>> flow_opts(["new"], True)
         Traceback (most recent call last):
-        cylc.flow.exceptions.InputError: --wait is not compatible ...
+        cylc.flow.exceptions.InputError: --wait is not compatible with
+        --flow=new or --flow=none
 
         >>> flow_opts(["new"], False, allow_new_or_none=False)
         Traceback (most recent call last):
         cylc.flow.exceptions.InputError: ... must be an integer, or 'all'
 
     """
+    if not flows:
+        return
+
+    flows = [val.strip() for val in flows]
+
     for val in flows:
         val = val.strip()
         if val in {FLOW_NONE, FLOW_NEW, FLOW_ALL}:
             if len(flows) != 1:
-                raise InputError(ERR_OPT_FLOW_INT)
+                raise InputError(ERR_OPT_FLOW_COMBINE.format(val))
             if not allow_new_or_none and val in {FLOW_NEW, FLOW_NONE}:
                 raise InputError(ERR_OPT_FLOW_VAL_2)
         else:
             try:
                 int(val)
             except ValueError:
-                raise InputError(ERR_OPT_FLOW_VAL.format(val)) from None
+                raise InputError(ERR_OPT_FLOW_VAL) from None
 
     if flow_wait and flows[0] in {FLOW_NEW, FLOW_NONE}:
         raise InputError(ERR_OPT_FLOW_WAIT)

cylc/flow/config.py

@@ -690,7 +690,7 @@ def process_initial_cycle_point(self) -> None:
         Sets:
             self.initial_point
             self.cfg['scheduling']['initial cycle point']
-            self.options.icp
+            self.evaluated_icp
         Raises:
             WorkflowConfigError - if it fails to validate
         """
@@ -710,10 +710,11 @@ def process_initial_cycle_point(self) -> None:
                     icp = ingest_time(orig_icp, get_current_time_string())
                 except IsodatetimeError as exc:
                     raise WorkflowConfigError(str(exc)) from None
-        if orig_icp != icp:
+        self.evaluated_icp = None
+        if icp != orig_icp:
             # now/next()/previous() was used, need to store
             # evaluated point in DB
-            self.options.icp = icp
+            self.evaluated_icp = icp
         self.initial_point = get_point(icp).standardise()
         self.cfg['scheduling']['initial cycle point'] = str(self.initial_point)
 

cylc/flow/data_store_mgr.py

@@ -1419,7 +1419,7 @@ def apply_task_proxy_db_history(self):
             itask, is_parent = self.db_load_task_proxies[relative_id]
             itask.submit_num = submit_num
             flow_nums = deserialise_set(flow_nums_str)
-            # Do not set states and outputs for future tasks in flow.
+            # Do not set states and outputs for inactive tasks in flow.
             if (
                     itask.flow_nums and
                     flow_nums != itask.flow_nums and
@@ -2250,7 +2250,9 @@ def delta_broadcast(self):
 
     def _generate_broadcast_node_deltas(self, node_data, node_type):
         cfg = self.schd.config.cfg
-        for node_id, node in node_data.items():
+        # NOTE: node_data may change during operation so make a copy
+        # see https://github.com/cylc/cylc-flow/pull/6397
+        for node_id, node in list(node_data.items()):
             tokens = Tokens(node_id)
             new_runtime = runtime_from_config(
                 self._apply_broadcasts_to_runtime(

cylc/flow/etc/tutorial/cylc-forecasting-workflow/lib/python/util.py

@@ -282,13 +282,18 @@ def __call__(self, grid_x, grid_y):
         return z_val
 
 
-def parse_domain(domain):
-    bbox = list(map(float, domain.split(',')))
+def parse_domain(domain: str):
+    lng1, lat1, lng2, lat2 = list(map(float, domain.split(',')))
+    msg = "Invalid domain '{}' ({} {} >= {})"
+    if lng1 >= lng2:
+        raise ValueError(msg.format(domain, 'longitude', lng1, lng2))
+    if lat1 >= lat2:
+        raise ValueError(msg.format(domain, 'latitude', lat1, lat2))
     return {
-        'lng1': bbox[0],
-        'lat1': bbox[1],
-        'lng2': bbox[2],
-        'lat2': bbox[3]
+        'lng1': lng1,
+        'lat1': lat1,
+        'lng2': lng2,
+        'lat2': lat2,
     }
 
 

cylc/flow/network/schema.py

@@ -2027,17 +2027,20 @@ class Arguments:
 class FlowMutationArguments:
     flow = graphene.List(
         graphene.NonNull(Flow),
-        default_value=[FLOW_ALL],
+        default_value=[],
         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:
-
-            * {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.
+            The flow(s) to trigger/set these tasks in.
+
+            By default:
+            * active tasks (n=0) keep their existing flow assignment
+            * inactive tasks (n>0) get assigned all active flows
+
+            Otherwise you can assign (inactive tasks) or add to (active tasks):
+            * a list of integer flow numbers
+            or one of the following strings:
+            * {FLOW_ALL} - all active flows
+            * {FLOW_NEW} - an automatically generated new flow number
+            * {FLOW_NONE} - (ignored for active tasks): no flow
         ''')
     )
     flow_wait = Boolean(

cylc/flow/scripts/trigger.py

@@ -17,19 +17,12 @@
 
 """cylc trigger [OPTIONS] ARGS
 
-Force tasks to run despite unsatisfied prerequisites.
+Force tasks to run regardless of prerequisites.
 
 * Triggering an unqueued waiting task queues it, regardless of prerequisites.
 * Triggering a queued task submits it, regardless of queue limiting.
 * Triggering an active task has no effect (it already triggered).
 
-Incomplete and active-waiting tasks in the n=0 window already belong to a flow.
-Triggering them queues them to run (or rerun) in the same flow.
-
-Beyond n=0, triggered tasks get all current active flow numbers by default, or
-specified flow numbers via the --flow option. Those flows - if/when they catch
-up - will see tasks that ran after triggering event as having run already.
-
 Examples:
   # trigger task foo in cycle 1234 in test
   $ cylc trigger test//1234/foo
@@ -39,6 +32,21 @@
 
   # start a new flow by triggering 1234/foo in test
   $ cylc trigger --flow=new test//1234/foo
+
+Flows:
+  Active tasks (in the n=0 window) already belong to a flow.
+  * by default, if triggered, they run in the same flow
+  * or with --flow=all, they are assigned all active flows
+  * or with --flow=INT or --flow=new, the original and new flows are merged
+  * (--flow=none is ignored for active tasks)
+
+  Inactive tasks (n>0) do not already belong to a flow.
+  * by default they are assigned all active flows
+  * otherwise, they are assigned the --flow value
+
+  Note --flow=new increments the global flow counter with each use. If it
+  takes multiple commands to start a new flow use the actual flow number
+  after the first command (you can read it from the scheduler log).
 """
 
 from functools import partial