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 {

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")

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)
+        # TODO: consider accumulating attribute statuses onto the families
         if fp_id in self.state_update_families:
             fp_updated = self.updated[FAMILY_PROXIES]
             tp_data = self.data[self.workflow_id][TASK_PROXIES]
@@ -2296,7 +2301,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 +2530,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 +2541,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

@@ -197,6 +197,7 @@ class SortArgs(InputObjectType):
     'sort': SortArgs(default_value=None),
 }
 
+# TODO: consider making these args searchable
 PROXY_ARGS = {
     'ids': graphene.List(ID, default_value=[]),
     'exids': graphene.List(ID, default_value=[]),
@@ -1124,6 +1125,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.',
     )

cylc/flow/network/schema.py

@@ -1294,7 +1294,12 @@ 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}")