Merge pull request #6708 from cylc/8.4.x-sync

MetRonnie · web-flow · commit 1ff9fb0fbb66 · 2025-04-07T15:13:33.000+01:00
🤖 Merge 8.4.x-sync into master
diff --git a/changes.d/6169.fix b/changes.d/6169.fix
@@ -0,0 +1 @@
+Ensure that job submit/failure is logged, even when retries are planned.
diff --git a/changes.d/6589.fix.md b/changes.d/6589.fix.md
@@ -0,0 +1 @@
+Fix potential accumulation of old families in UI.
diff --git a/cylc/flow/data_store_mgr.py b/cylc/flow/data_store_mgr.py
@@ -1920,10 +1920,14 @@ def _family_ascent_point_prune(
         family. The work back up to origin checking these families are active.
 
         """
-        fp_data = self.data[self.workflow_id][FAMILY_PROXIES]
         fp_updated = self.updated[FAMILY_PROXIES]
-        if fp_id in fp_data:
-            fam_node = fp_data[fp_id]
+        fam_node = self.data[self.workflow_id][FAMILY_PROXIES].get(
+            fp_id,
+            self.added[FAMILY_PROXIES].get(fp_id, None)
+        )
+        # Should never be None,
+        # leaving in as protection against potential race conditions.
+        if fam_node is not None:
             # Gather child families, then check/update recursively
             for child_id in fam_node.child_families:
                 if child_id in checked_ids:
@@ -1938,11 +1942,14 @@ def _family_ascent_point_prune(
                     child_tasks.update(fp_updated[fp_id].child_tasks)
                 if fp_updated[fp_id].child_families:
                     child_families.update(fp_updated[fp_id].child_families)
-            # if any child tasks or families are in window, don't prune.
+            # If any child tasks or families are in window,
+            # then don't prune this family.
             if (
                 child_tasks.difference(node_ids)
                 or child_families.difference(prune_ids)
             ):
+                # If any child tasks or families will be pruned,
+                # then update family states.
                 if (
                     child_tasks.intersection(node_ids)
                     or child_families.intersection(prune_ids)
diff --git a/cylc/flow/task_events_mgr.py b/cylc/flow/task_events_mgr.py
@@ -749,7 +749,7 @@ def process_message(
                 # Already failed.
                 return True
             if self._process_message_failed(
-                itask, event_time, self.JOB_FAILED, forced
+                itask, event_time, self.JOB_FAILED, forced, message
             ):
                 self.spawn_children(itask, TASK_OUTPUT_FAILED)
 
@@ -798,7 +798,7 @@ def process_message(
             self.workflow_db_mgr.put_update_task_jobs(
                 itask, {"run_signal": signal})
             if self._process_message_failed(
-                itask, event_time, self.JOB_FAILED, forced
+                itask, event_time, self.JOB_FAILED, forced, message
             ):
                 self.spawn_children(itask, TASK_OUTPUT_FAILED)
 
@@ -815,7 +815,7 @@ def process_message(
             self.workflow_db_mgr.put_update_task_jobs(
                 itask, {"run_signal": aborted_with})
             if self._process_message_failed(
-                itask, event_time, aborted_with, forced
+                itask, event_time, aborted_with, forced, message
             ):
                 self.spawn_children(itask, TASK_OUTPUT_FAILED)
 
@@ -930,11 +930,15 @@ def _process_message_check(
             return False
 
         severity_lvl: int = LOG_LEVELS.get(severity, INFO)
+        # Don't log submit/failure messages here:
+        if flag != self.FLAG_POLLED and message in {
+            self.EVENT_SUBMIT_FAILED, f'{FAIL_MESSAGE_PREFIX}ERR'
+        }:
+            return True
         # Demote log level to DEBUG if this is a message that duplicates what
         # gets logged by itask state change anyway (and not manual poll)
         if severity_lvl > DEBUG and flag != self.FLAG_POLLED and message in {
             self.EVENT_SUBMITTED, self.EVENT_STARTED, self.EVENT_SUCCEEDED,
-            self.EVENT_SUBMIT_FAILED, f'{FAIL_MESSAGE_PREFIX}ERR'
         }:
             severity_lvl = DEBUG
         LOG.log(severity_lvl, f"[{itask}] {flag}{message}{timestamp}")
@@ -1305,10 +1309,16 @@ def _process_message_failed(
         event_time: Optional[str],
         message: str,
         forced: bool,
+        full_message: str,
     ) -> bool:
         """Helper for process_message, handle a failed message.
 
         Return True if no retries (hence go to the failed state).
+
+        Args:
+            full_message:
+                If we have retries lined up we still tell users what
+                happened to cause the this attempt to fail.
         """
         no_retries = False
         if event_time is None:
@@ -1321,6 +1331,7 @@ def _process_message_failed(
             "run_status": 1,
             "time_run_exit": event_time,
         })
+        LOG.error(f'[{itask}] {full_message or self.EVENT_FAILED}')
         if (
             forced
             or TimerFlags.EXECUTION_RETRY not in itask.try_timers
@@ -1345,7 +1356,7 @@ def _process_message_failed(
             timer = itask.try_timers[TimerFlags.EXECUTION_RETRY]
             self._retry_task(itask, timer.timeout)
             delay_msg = f"retrying in {timer.delay_timeout_as_str()}"
-            LOG.warning(f"[{itask}] {delay_msg}")
+            LOG.warning(f'[{itask}] - {delay_msg}')
             msg = f"{self.JOB_FAILED}, {delay_msg}"
             self.setup_event_handlers(itask, self.EVENT_RETRY, msg)
         self._reset_job_timers(itask)
@@ -1423,14 +1434,14 @@ def _process_message_submit_failed(
         Return True if no retries (hence go to the submit-failed state).
         """
         no_retries = False
-        LOG.critical(f"[{itask}] {self.EVENT_SUBMIT_FAILED}")
         if event_time is None:
             event_time = get_current_time_string()
         self.workflow_db_mgr.put_update_task_jobs(itask, {
             "time_submit_exit": event_time,
             "submit_status": 1,
         })
         itask.summary['submit_method_id'] = None
+        LOG.error(f"[{itask}] {self.EVENT_SUBMIT_FAILED}")
         if (
             forced
             or TimerFlags.SUBMISSION_RETRY not in itask.try_timers
diff --git a/tests/functional/cylc-remove/00-simple/flow.cylc b/tests/functional/cylc-remove/00-simple/flow.cylc
@@ -17,7 +17,7 @@
         script = false
     [[cleaner]]
         script = """
-            cylc__job__poll_grep_workflow_log -E '1/b/01:running.* \(received\)failed'
+            cylc__job__poll_grep_workflow_log -E '1/b/01.* failed'
             # Remove the unhandled failed task
             cylc remove "$CYLC_WORKFLOW_ID//1/b"
             # Remove waiting 1/c
diff --git a/tests/functional/cylc-remove/02-cycling/flow.cylc b/tests/functional/cylc-remove/02-cycling/flow.cylc
@@ -17,8 +17,8 @@
 [runtime]
     [[remover]]
         script = """
-            cylc__job__poll_grep_workflow_log -E '2020/bar/01:running.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '2021/baz/01:running.* \(received\)failed'
+            cylc__job__poll_grep_workflow_log -E '2020/bar/01.* failed'
+            cylc__job__poll_grep_workflow_log -E '2021/baz/01.* failed'
             # Remove the two unhandled failed tasks.
             cylc remove "$CYLC_WORKFLOW_ID//*/ba*:failed"
             # Remove the two unsatisfied waiting tasks.
diff --git a/tests/functional/cylc-trigger/02-filter-failed/flow.cylc b/tests/functional/cylc-trigger/02-filter-failed/flow.cylc
@@ -18,9 +18,9 @@
     [[fixer]]
         script = """
             cylc__job__wait_cylc_message_started
-            cylc__job__poll_grep_workflow_log -E '1/fixable1/01:running.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable2/01:running.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable3/01:running.* \(received\)failed'
+            cylc__job__poll_grep_workflow_log -E '\[1/fixable1/01:running\] failed/ERR'
+            cylc__job__poll_grep_workflow_log -E '\[1/fixable2/01:running\] failed/ERR'
+            cylc__job__poll_grep_workflow_log -E '\[1/fixable3/01:running\] failed/ERR'
             cylc trigger "${CYLC_WORKFLOW_ID}//1/fixable*"
         """
     [[Z]]
diff --git a/tests/functional/cylc-trigger/04-filter-names/flow.cylc b/tests/functional/cylc-trigger/04-filter-names/flow.cylc
@@ -22,11 +22,11 @@
     [[fixer]]
         script = """
             cylc__job__wait_cylc_message_started
-            cylc__job__poll_grep_workflow_log -E '1/fixable-1a/01.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable-1b/01.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable-2a/01.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable-2b/01.* \(received\)failed'
-            cylc__job__poll_grep_workflow_log -E '1/fixable-3/01.* \(received\)failed'
+            cylc__job__poll_grep_workflow_log -E '1/fixable-1a/01.* failed'
+            cylc__job__poll_grep_workflow_log -E '1/fixable-1b/01.* failed'
+            cylc__job__poll_grep_workflow_log -E '1/fixable-2a/01.* failed'
+            cylc__job__poll_grep_workflow_log -E '1/fixable-2b/01.* failed'
+            cylc__job__poll_grep_workflow_log -E '1/fixable-3/01.* failed'
             cylc trigger "${CYLC_WORKFLOW_ID}//" \
                 '//1/FIXABLE-1' '//1/fixable-2*' '//1/fixable-3'
         """
diff --git a/tests/functional/hold-release/11-retrying/flow.cylc b/tests/functional/hold-release/11-retrying/flow.cylc
@@ -18,7 +18,7 @@ t-retry-able => t-analyse
     [[t-hold-release]]
         script = """
             cylc__job__poll_grep_workflow_log -E \
-                '1/t-retry-able/01:running.* \(received\)failed'
+                '\[1/t-retry-able:waiting\] - retrying'
 
             cylc__job__poll_grep_workflow_log -E \
                 '1/t-retry-able/01:running.* => waiting'
diff --git a/tests/functional/reload/25-xtriggers.t b/tests/functional/reload/25-xtriggers.t
@@ -42,7 +42,7 @@ init_workflow "${TEST_NAME_BASE}" <<'__FLOW_CONFIG__'
     [[reload]]
         script = """
             # wait for "broken" to fail
-            cylc__job__poll_grep_workflow_log -E '1/broken/01.* \(received\)failed/ERR'
+            cylc__job__poll_grep_workflow_log -E '1/broken/01.*failed/ERR'
             # fix "broken" to allow it to pass
             sed -i 's/false/true/' "${CYLC_WORKFLOW_RUN_DIR}/flow.cylc"
             # reload the workflow
@@ -63,7 +63,7 @@ workflow_run_ok "${TEST_NAME_BASE}-run" cylc play "${WORKFLOW_NAME}" --no-detach
 log_scan "${TEST_NAME_BASE}-scan" \
     "$(cylc cat-log -m p "${WORKFLOW_NAME}")" \
     1 1 \
-    '1/broken.* (received)failed/ERR'
+    '1/broken/01.*failed/ERR'
 
 log_scan "${TEST_NAME_BASE}-scan" \
     "$(cylc cat-log -m p "${WORKFLOW_NAME}")" 1 1 \
diff --git a/tests/functional/reload/runahead/flow.cylc b/tests/functional/reload/runahead/flow.cylc
@@ -20,7 +20,7 @@
         script = true
     [[reloader]]
         script = """
-            cylc__job__poll_grep_workflow_log -E "${CYLC_TASK_CYCLE_POINT}/foo/01:running.*\(received\)failed"
+            cylc__job__poll_grep_workflow_log -E "${CYLC_TASK_CYCLE_POINT}/foo/01:running.*failed"
             perl -pi -e 's/(runahead limit = )P1( # marker)/\1 P3\2/' $CYLC_WORKFLOW_RUN_DIR/flow.cylc
             cylc reload $CYLC_WORKFLOW_ID
         """
diff --git a/tests/functional/spawn-on-demand/10-retrigger/flow.cylc b/tests/functional/spawn-on-demand/10-retrigger/flow.cylc
@@ -18,7 +18,7 @@
       """
    [[triggerer]]
       script = """
-         cylc__job__poll_grep_workflow_log -E '1/oops/01:running.* \(received\)failed'
+         cylc__job__poll_grep_workflow_log -E '1/oops/01.* failed'
          cylc trigger "${CYLC_WORKFLOW_ID}//1/oops"
        """
    [[foo, bar]]
diff --git a/tests/functional/triggering/19-and-suicide/flow.cylc b/tests/functional/triggering/19-and-suicide/flow.cylc
@@ -16,7 +16,7 @@
     [[t0]]
         # https://github.com/cylc/cylc-flow/issues/2655
         # "1/t2" should not suicide on "1/t1:failed"
-        script = cylc__job__poll_grep_workflow_log -E '1/t1.* \(received\)failed'
+        script = cylc__job__poll_grep_workflow_log -E '1/t1.* failed'
     [[t1]]
         script = false
     [[t2]]
diff --git a/tests/integration/test_data_store_mgr.py b/tests/integration/test_data_store_mgr.py
@@ -318,10 +318,123 @@ async def test_update_data_structure(harness):
     assert TASK_STATUS_FAILED in set(collect_states(data, FAMILY_PROXIES))
     # state totals changed
     assert TASK_STATUS_FAILED in data[WORKFLOW].state_totals
-    # Shows pruning worked
-    # TODO: fixme
-    # https://github.com/cylc/cylc-flow/issues/4175#issuecomment-1025666413
-    # assert len({t.is_held for t in data[TASK_PROXIES].values()}) == 1
+
+
+async def test_prune_data_store(flow, scheduler, start):
+    """Test prune_data_store. This method will expand and reduce the data-store
+    to invoke pruning.
+
+    Also test rapid addition and removal of families (as happens with suicide
+    triggers):
+    https://github.com/cylc/cylc-ui/issues/1999
+
+    """
+    id_ = flow({
+        'scheduling': {
+            'graph': {
+                'R1': 'foo => bar'
+            }
+        },
+        'runtime': {
+            'FOOBAR': {},
+            'FOO': {
+                'inherit': 'FOOBAR'
+            },
+            'foo': {
+                'inherit': 'FOO'
+            },
+            'BAR': {
+                'inherit': 'FOOBAR'
+            },
+            'bar': {
+                'inherit': 'BAR'
+            }
+        }
+    })
+    schd = scheduler(id_)
+    async with start(schd):
+        # initialise the data store
+        await schd.update_data_structure()
+        w_id = schd.data_store_mgr.workflow_id
+        data = schd.data_store_mgr.data[w_id]
+        schd.pool.hold_tasks(['*'])
+        await schd.update_data_structure()
+        assert len({t.is_held for t in data[TASK_PROXIES].values()}) == 2
+
+        # Window size reduction to invoke pruning
+        schd.data_store_mgr.set_graph_window_extent(0)
+        schd.data_store_mgr.update_data_structure()
+        assert len({t.is_held for t in data[TASK_PROXIES].values()}) == 1
+
+        # Test rapid addition and removal
+        # bar/BAR task/family proxies not in .added
+        assert len({
+            t.name
+            for t in schd.data_store_mgr.added[TASK_PROXIES].values()
+            if t.name == 'bar'
+        }) == 0
+        assert len({
+            f.name
+            for f in schd.data_store_mgr.added[FAMILY_PROXIES].values()
+            if f.name == 'BAR'
+        }) == 0
+        # Add bar/BAR on set output of foo
+        for itask in schd.pool.get_tasks():
+            schd.pool.spawn_on_output(itask, TASK_OUTPUT_SUCCEEDED)
+        # bar/BAR now found.
+        assert len({
+            t.name
+            for t in schd.data_store_mgr.added[TASK_PROXIES].values()
+            if t.name == 'bar'
+        }) == 1
+        assert len({
+            f.name
+            for f in schd.data_store_mgr.added[FAMILY_PROXIES].values()
+            if f.name == 'BAR'
+        }) == 1
+        # Before updating the data-store, remove bar/BAR.
+        schd.pool.remove(schd.pool._get_task_by_id('1/bar'), 'Test removal')
+        schd.data_store_mgr.update_data_structure()
+        # bar/BAR not found in data or added stores.
+        assert len({
+            t.name
+            for t in data[TASK_PROXIES].values()
+            if t.name == 'bar'
+        }) == 0
+        assert len({
+            t.name
+            for t in schd.data_store_mgr.added[TASK_PROXIES].values()
+            if t.name == 'bar'
+        }) == 0
+        assert len({
+            f.name
+            for f in data[FAMILY_PROXIES].values()
+            if f.name == 'BAR'
+        }) == 0
+        assert len({
+            f.name
+            for f in schd.data_store_mgr.added[FAMILY_PROXIES].values()
+            if f.name == 'BAR'
+        }) == 0
+
+
+async def test_family_ascent_point_prune(harness):
+    """Test _family_ascent_point_prune. This method tries to remove
+    non-existent family."""
+    schd, data = harness
+    fp_id = 'NotAFamilyProxy'
+    parent_ids = {fp_id}
+    checked_ids = set()
+    node_ids = set()
+    schd.data_store_mgr._family_ascent_point_prune(
+        next(iter(parent_ids)),
+        node_ids,
+        parent_ids,
+        checked_ids,
+        schd.data_store_mgr.family_pruned_ids
+    )
+    assert len(checked_ids) == 1
+    assert len(parent_ids) == 0
 
 
 def test_delta_task_prerequisite(harness):
diff --git a/tests/integration/test_task_events_mgr.py b/tests/integration/test_task_events_mgr.py
diff --git a/tests/integration/test_task_pool.py b/tests/integration/test_task_pool.py
diff --git a/tests/integration/test_workflow_events.py b/tests/integration/test_workflow_events.py
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
diff --git a/tests/unit/test_taskdef.py b/tests/unit/test_taskdef.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+Ensure that job submit/failure is logged, even when retries are planned.`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+Fix potential accumulation of old families in UI.`