xtriggers: fix update and add is_retry, is_wallclock and is_xtriggered attrs

* Don't update xtriggers incrementally:
  * Xtriggers were being updated in the data store incrementally.
  * This doesn't work with GraphQL subscriptions
    (see cylc#6307).
  * This turns off incremental updates ensuring that all xtriggers are
    included in all updates.
  * Unblocks cylc/cylc-ui#2103
* Add is_retry, is_wallclock and is_xtriggered task attributes.
  * We would like to be able to incorporate xtriggers into the task
    icons used in the GUI and Tui.
  * These attributes allow clients to access the required information
    without having to subscribe to and monitor all xtriggers.
  * Unblocks cylc/cylc-ui#331

Author: oliver-sanders

cylc/flow/config.py

@@ -1937,7 +1937,7 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
                 )
 
             self.xtrigger_collator.add_trig(label, xtrig, self.fdir)
-            self.taskdefs[right].add_xtrig_label(label, seq)
+            self.taskdefs[right].add_xtrig_label(label, seq, xtrig.func_name)
 
     def get_actual_first_point(self, start_point):
         """Get actual first cycle point for the workflow
@@ -2679,4 +2679,4 @@ def upgrade_clock_triggers(self):
             # Add it to the task, for each sequence that the task appears in.
             taskdef = self.get_taskdef(task_name)
             for seq in taskdef.sequences:
-                taskdef.add_xtrig_label(label, seq)
+                taskdef.add_xtrig_label(label, seq, xtrig.func_name)

cylc/flow/data_messages.proto

@@ -231,6 +231,9 @@ message PbTaskProxy {
     optional bool flow_wait = 27;
     optional PbRuntime runtime = 28;
     optional int32 graph_depth = 29;
+    optional bool is_retry = 30;
+    optional bool is_wallclock = 31;
+    optional bool is_xtriggered = 32;
 }
 
 message PbFamily {
@@ -270,6 +273,9 @@ message PbFamilyProxy {
     optional int32 is_runahead_total = 20;
     optional PbRuntime runtime = 21;
     optional int32 graph_depth = 22;
+    optional bool is_retry = 23;
+    optional bool is_wallclock = 24;
+    optional bool is_xtriggered = 25;
 }
 
 message PbEdge {

cylc/flow/data_messages_pb2.py

@@ -304,7 +304,7 @@ class PbTrigger(_message.Message):
     def __init__(self, id: _Optional[str] = ..., label: _Optional[str] = ..., message: _Optional[str] = ..., satisfied: bool = ..., time: _Optional[float] = ...) -> None: ...
 
 class PbTaskProxy(_message.Message):
-    __slots__ = ("stamp", "id", "task", "state", "cycle_point", "depth", "job_submits", "outputs", "namespace", "prerequisites", "jobs", "first_parent", "name", "is_held", "edges", "ancestors", "flow_nums", "external_triggers", "xtriggers", "is_queued", "is_runahead", "flow_wait", "runtime", "graph_depth")
+    __slots__ = ("stamp", "id", "task", "state", "cycle_point", "depth", "job_submits", "outputs", "namespace", "prerequisites", "jobs", "first_parent", "name", "is_held", "edges", "ancestors", "flow_nums", "external_triggers", "xtriggers", "is_queued", "is_runahead", "flow_wait", "runtime", "graph_depth", "is_retry", "is_wallclock", "is_xtriggered")
     class OutputsEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -350,6 +350,9 @@ class PbTaskProxy(_message.Message):
     FLOW_WAIT_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_FIELD_NUMBER: _ClassVar[int]
     GRAPH_DEPTH_FIELD_NUMBER: _ClassVar[int]
+    IS_RETRY_FIELD_NUMBER: _ClassVar[int]
+    IS_WALLCLOCK_FIELD_NUMBER: _ClassVar[int]
+    IS_XTRIGGERED_FIELD_NUMBER: _ClassVar[int]
     stamp: str
     id: str
     task: str
@@ -374,7 +377,10 @@ class PbTaskProxy(_message.Message):
     flow_wait: bool
     runtime: PbRuntime
     graph_depth: int
-    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., task: _Optional[str] = ..., state: _Optional[str] = ..., cycle_point: _Optional[str] = ..., depth: _Optional[int] = ..., job_submits: _Optional[int] = ..., outputs: _Optional[_Mapping[str, PbOutput]] = ..., namespace: _Optional[_Iterable[str]] = ..., prerequisites: _Optional[_Iterable[_Union[PbPrerequisite, _Mapping]]] = ..., jobs: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., name: _Optional[str] = ..., is_held: bool = ..., edges: _Optional[_Iterable[str]] = ..., ancestors: _Optional[_Iterable[str]] = ..., flow_nums: _Optional[str] = ..., external_triggers: _Optional[_Mapping[str, PbTrigger]] = ..., xtriggers: _Optional[_Mapping[str, PbTrigger]] = ..., is_queued: bool = ..., is_runahead: bool = ..., flow_wait: bool = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ...) -> None: ...
+    is_retry: bool
+    is_wallclock: bool
+    is_xtriggered: bool
+    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., task: _Optional[str] = ..., state: _Optional[str] = ..., cycle_point: _Optional[str] = ..., depth: _Optional[int] = ..., job_submits: _Optional[int] = ..., outputs: _Optional[_Mapping[str, PbOutput]] = ..., namespace: _Optional[_Iterable[str]] = ..., prerequisites: _Optional[_Iterable[_Union[PbPrerequisite, _Mapping]]] = ..., jobs: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., name: _Optional[str] = ..., is_held: bool = ..., edges: _Optional[_Iterable[str]] = ..., ancestors: _Optional[_Iterable[str]] = ..., flow_nums: _Optional[str] = ..., external_triggers: _Optional[_Mapping[str, PbTrigger]] = ..., xtriggers: _Optional[_Mapping[str, PbTrigger]] = ..., is_queued: bool = ..., is_runahead: bool = ..., flow_wait: bool = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ..., is_retry: bool = ..., is_wallclock: bool = ..., is_xtriggered: bool = ...) -> None: ...
 
 class PbFamily(_message.Message):
     __slots__ = ("stamp", "id", "name", "meta", "depth", "proxies", "parents", "child_tasks", "child_families", "first_parent", "runtime", "descendants")
@@ -405,7 +411,7 @@ class PbFamily(_message.Message):
     def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., name: _Optional[str] = ..., meta: _Optional[_Union[PbMeta, _Mapping]] = ..., depth: _Optional[int] = ..., proxies: _Optional[_Iterable[str]] = ..., parents: _Optional[_Iterable[str]] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., descendants: _Optional[_Iterable[str]] = ...) -> None: ...
 
 class PbFamilyProxy(_message.Message):
-    __slots__ = ("stamp", "id", "cycle_point", "name", "family", "state", "depth", "first_parent", "child_tasks", "child_families", "is_held", "ancestors", "states", "state_totals", "is_held_total", "is_queued", "is_queued_total", "is_runahead", "is_runahead_total", "runtime", "graph_depth")
+    __slots__ = ("stamp", "id", "cycle_point", "name", "family", "state", "depth", "first_parent", "child_tasks", "child_families", "is_held", "ancestors", "states", "state_totals", "is_held_total", "is_queued", "is_queued_total", "is_runahead", "is_runahead_total", "runtime", "graph_depth", "is_retry", "is_wallclock", "is_xtriggered")
     class StateTotalsEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -434,6 +440,9 @@ class PbFamilyProxy(_message.Message):
     IS_RUNAHEAD_TOTAL_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_FIELD_NUMBER: _ClassVar[int]
     GRAPH_DEPTH_FIELD_NUMBER: _ClassVar[int]
+    IS_RETRY_FIELD_NUMBER: _ClassVar[int]
+    IS_WALLCLOCK_FIELD_NUMBER: _ClassVar[int]
+    IS_XTRIGGERED_FIELD_NUMBER: _ClassVar[int]
     stamp: str
     id: str
     cycle_point: str
@@ -455,7 +464,10 @@ class PbFamilyProxy(_message.Message):
     is_runahead_total: int
     runtime: PbRuntime
     graph_depth: int
-    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., cycle_point: _Optional[str] = ..., name: _Optional[str] = ..., family: _Optional[str] = ..., state: _Optional[str] = ..., depth: _Optional[int] = ..., first_parent: _Optional[str] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., is_held: bool = ..., ancestors: _Optional[_Iterable[str]] = ..., states: _Optional[_Iterable[str]] = ..., state_totals: _Optional[_Mapping[str, int]] = ..., is_held_total: _Optional[int] = ..., is_queued: bool = ..., is_queued_total: _Optional[int] = ..., is_runahead: bool = ..., is_runahead_total: _Optional[int] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ...) -> None: ...
+    is_retry: bool
+    is_wallclock: bool
+    is_xtriggered: bool
+    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., cycle_point: _Optional[str] = ..., name: _Optional[str] = ..., family: _Optional[str] = ..., state: _Optional[str] = ..., depth: _Optional[int] = ..., first_parent: _Optional[str] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., is_held: bool = ..., ancestors: _Optional[_Iterable[str]] = ..., states: _Optional[_Iterable[str]] = ..., state_totals: _Optional[_Mapping[str, int]] = ..., is_held_total: _Optional[int] = ..., is_queued: bool = ..., is_queued_total: _Optional[int] = ..., is_runahead: bool = ..., is_runahead_total: _Optional[int] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ..., is_retry: bool = ..., is_wallclock: bool = ..., is_xtriggered: bool = ...) -> None: ...
 
 class PbEdge(_message.Message):
     __slots__ = ("stamp", "id", "source", "target", "suicide", "cond")

cylc/flow/data_messages_pb2.pyi

@@ -1484,6 +1484,17 @@ def apply_task_proxy_db_history(self):
 
         self.db_load_task_proxies.clear()
 
+    def _populate_xtriggers(self, itask, tproxy):
+        """Transfer xtriggers from the itask onto the PbTaskProxy."""
+        for label, (satisfied, _) in itask.state.xtriggers.items():
+            sig = self.schd.xtrigger_mgr.get_xtrig_ctx(
+                itask, label).get_signature()
+            xtrig = tproxy.xtriggers[sig]
+            xtrig.id = sig
+            xtrig.label = label
+            xtrig.satisfied = satisfied
+            self.xtrigger_tasks.setdefault(sig, set()).add((tproxy.id, label))
+
     def _process_internal_task_proxy(
         self,
         itask: 'TaskProxy',
@@ -1517,14 +1528,7 @@ def _process_internal_task_proxy(
             ext_trig.id = trig
             ext_trig.satisfied = satisfied
 
-        for label, satisfied in itask.state.xtriggers.items():
-            sig = self.schd.xtrigger_mgr.get_xtrig_ctx(
-                itask, label).get_signature()
-            xtrig = tproxy.xtriggers[sig]
-            xtrig.id = sig
-            xtrig.label = label
-            xtrig.satisfied = satisfied
-            self.xtrigger_tasks.setdefault(sig, set()).add((tproxy.id, label))
+        self._populate_xtriggers(itask, tproxy)
 
         if tproxy.state in self.latest_state_tasks:
             tp_ref = itask.identity
@@ -2034,6 +2038,7 @@ def _family_ascent_point_update(self, fp_id):
             if child_fam_id in self.updated_state_families:
                 continue
             self._family_ascent_point_update(child_fam_id)
+
         if fp_id in self.state_update_families:
             fp_updated = self.updated[FAMILY_PROXIES]
             tp_data = self.data[self.workflow_id][TASK_PROXIES]
@@ -2044,6 +2049,9 @@ def _family_ascent_point_update(self, fp_id):
             is_held_total = 0
             is_queued_total = 0
             is_runahead_total = 0
+            is_retry = False
+            is_wallclock = False
+            is_xtriggered = False
             graph_depth = self.n_edge_distance
             for child_id in fam_node.child_families:
                 child_node = fp_updated.get(child_id, fp_data.get(child_id))
@@ -2082,12 +2090,38 @@ def _family_ascent_point_update(self, fp_id):
                     is_queued_total += 1
 
                 tp_runahead = tp_delta
-                if (tp_runahead is None
-                        or not tp_runahead.HasField('is_runahead')):
+                if (
+                    tp_runahead is None
+                    or not tp_runahead.HasField('is_runahead')
+                ):
                     tp_runahead = tp_node
                 if tp_runahead.is_runahead:
                     is_runahead_total += 1
 
+                tp_retry = tp_delta
+                if tp_retry is None or not tp_retry.HasField('is_retry'):
+                    tp_retry = tp_node
+                if tp_retry.is_retry:
+                    is_retry = True
+
+                tp_wallclock = tp_delta
+                if (
+                    tp_wallclock is None
+                    or not tp_wallclock.HasField('is_wallclock')
+                ):
+                    tp_wallclock = tp_node
+                if tp_wallclock.is_wallclock:
+                    is_wallclock = True
+
+                tp_xtriggered = tp_delta
+                if (
+                    tp_xtriggered is None
+                    or not tp_xtriggered.HasField('is_xtriggered')
+                ):
+                    tp_xtriggered = tp_node
+                if tp_xtriggered.is_xtriggered:
+                    is_xtriggered = True
+
                 tp_depth = tp_delta
                 if tp_depth is None or not tp_depth.HasField('graph_depth'):
                     tp_depth = tp_node
@@ -2106,7 +2140,10 @@ def _family_ascent_point_update(self, fp_id):
                 is_queued_total=is_queued_total,
                 is_runahead=(is_runahead_total > 0),
                 is_runahead_total=is_runahead_total,
-                graph_depth=graph_depth
+                is_retry=is_retry,
+                is_wallclock=is_wallclock,
+                is_xtriggered=is_xtriggered,
+                graph_depth=graph_depth,
             )
             fp_delta.states[:] = state_counter.keys()
             # Use all states to clean up pruned counts
@@ -2296,7 +2333,14 @@ def delta_task_state(self, itask: 'TaskProxy') -> None:
             tp_id, PbTaskProxy(id=tp_id)
         )
         tp_delta.stamp = f'{tp_id}@{update_time}'
-        for field in ('is_held', 'is_queued', 'is_runahead'):
+        for field in (
+            'is_held',
+            'is_queued',
+            'is_runahead',
+            'is_retry',
+            'is_wallclock',
+            'is_xtriggered',
+        ):
             val = getattr(itask.state, field)
             if (
                 # only update the fields that have changed compared to store:
@@ -2518,7 +2562,7 @@ def delta_task_ext_trigger(
         ext_trigger.time = update_time
         self.updates_pending = True
 
-    def delta_task_xtrigger(self, sig, satisfied):
+    def delta_task_xtrigger(self, itask):
         """Create delta for change in task proxy xtrigger.
 
         Args:
@@ -2529,18 +2573,15 @@ def delta_task_xtrigger(self, sig, satisfied):
             satisfied (bool): Trigger message.
 
         """
-        update_time = time()
-        for tp_id, label in self.xtrigger_tasks.get(sig, set()):
-            # update task instance
-            tp_delta = self.updated[TASK_PROXIES].setdefault(
-                tp_id, PbTaskProxy(id=tp_id))
-            tp_delta.stamp = f'{tp_id}@{update_time}'
-            xtrigger = tp_delta.xtriggers[sig]
-            xtrigger.id = sig
-            xtrigger.label = label
-            xtrigger.satisfied = satisfied
-            xtrigger.time = update_time
-            self.updates_pending = True
+        tp_id, tproxy = self.store_node_fetcher(itask.tokens)
+        if not tproxy:
+            return
+
+        # refresh all xtriggers (we can't support incremental update on this
+        # field at present see https://github.com/cylc/cylc-flow/issues/6307)
+        self._populate_xtriggers(itask, tproxy)
+
+        self.updates_pending = True
 
     def delta_from_task_proxy(self, itask: TaskProxy) -> None:
         """Create delta from existing pool task proxy.

cylc/flow/data_store_mgr.py

@@ -1124,6 +1124,18 @@ class Meta:
     is_runahead = Boolean(
         description='True if this task is held back by the "runahead limit".',
     )
+    is_retry = Boolean(
+        description='True if this task has a scheduled retry.',
+    )
+    is_wallclock = Boolean(
+        description='True if this task has an unsatisfied wallclock trigger.',
+    )
+    is_xtriggered = Boolean(
+        description=sstrip(
+            'True if this task has an unsatisfied xtrigger'
+            ' (excluding retry and wallclock xtriggers).'
+        ),
+    )
     flow_nums = String(
         description='The flows this task instance belongs to.',
     )
@@ -1304,6 +1316,23 @@ class Meta:
     is_queued_total = Int()
     is_runahead = Boolean()
     is_runahead_total = Int()
+    is_retry = Boolean(
+        description=(
+            'True if this family contains a task that has a scheduled retry.'
+        ),
+    )
+    is_wallclock = Boolean(
+        description=(
+            'True if this family contains a task that has an'
+            ' unsatisfied wallclock trigger.'
+        ),
+    )
+    is_xtriggered = Boolean(
+        description=sstrip(
+            'True if this family contains a task that has an unsatisfied'
+            ' xtrigger (excluding retry and wallclock xtriggers).'
+        ),
+    )
     depth = Int()
     graph_depth = Int(
         description=sstrip('''

cylc/flow/network/schema.py

@@ -1280,7 +1280,7 @@ def _retry_task(self, itask, wallclock_time, submit_retry=False):
         if label in itask.state.xtriggers:
             # retry xtrigger already exists from a previous retry, modify it
             self.xtrigger_mgr.mutate_trig(label, kwargs)
-            itask.state.xtriggers[label] = False
+            itask.state.update_xtrigger(label, False)
         else:
             # create a new retry xtrigger
             xtrig = SubFuncContext(
@@ -1294,7 +1294,16 @@ def _retry_task(self, itask, wallclock_time, submit_retry=False):
                 xtrig,
                 os.getenv("CYLC_WORKFLOW_RUN_DIR")
             )
-            itask.state.add_xtrigger(label)
+            itask.state.add_xtrigger(label, label, False)
+
+        sig = label
+
+        (
+            self.data_store_mgr.xtrigger_tasks
+            .setdefault(sig, set())
+            .add((itask.identity, label))
+        )
+        self.data_store_mgr.delta_task_xtrigger(itask)
 
         if itask.state_reset(TASK_STATUS_WAITING):
             self.data_store_mgr.delta_task_state(itask)

cylc/flow/task_events_mgr.py

@@ -772,7 +772,7 @@ def get_or_spawn_task(
                 xtrig_label in (
                     self.xtrigger_mgr.xtriggers.sequential_xtrigger_labels)
                 for sequence, xtrig_labels in tdef.xtrig_labels.items()
-                for xtrig_label in xtrig_labels
+                for (xtrig_label, _) in xtrig_labels
                 if sequence.is_valid(point)
             ):
                 is_xtrig_sequential = True

cylc/flow/task_pool.py

@@ -541,8 +541,13 @@ def merge_flows(self, flow_nums: Set) -> None:
         )
 
     def state_reset(
-        self, status=None, is_held=None, is_queued=None, is_runahead=None,
-        silent=False, forced=False
+        self,
+        status=None,
+        is_held=None,
+        is_queued=None,
+        is_runahead=None,
+        silent=False,
+        forced=False
     ) -> bool:
         """Set new state and log the change. Return whether it changed.
 
@@ -554,7 +559,7 @@ def state_reset(
             is_runahead = False
 
         if self.state.reset(
-            status, is_held, is_queued, is_runahead, forced
+            status, is_held, is_queued, is_runahead, forced=forced
         ):
             if not silent and not self.transient:
                 LOG.info(f"[{before}] => {self.state}")