Fix expire trigger.

Update change log.
Add new func test.

CHANGES.md

@@ -17,8 +17,11 @@ ones in. -->
 [5398](https://github.com/cylc/cylc-flow/pull/5398) - Fix platform from
 group selection order bug.
 
+[#5412](https://github.com/cylc/cylc-flow/pull/5412) -
+Fix task expire trigger.
+
 [#5384](https://github.com/cylc/cylc-flow/pull/5384) -
-Fixes `cylc set-verbosity`.
+Fix `cylc set-verbosity`.
 
 [#5386](https://github.com/cylc/cylc-flow/pull/5386) Fix bug where
 absence of `job name length maximum` in PBS platform settings would cause

cylc/flow/task_pool.py

@@ -686,6 +686,7 @@ def remove(self, itask, reason=""):
             if not self.hidden_pool[itask.point]:
                 del self.hidden_pool[itask.point]
             LOG.debug(f"[{itask}] {msg}")
+            self.task_queue_mgr.remove_task(itask)
             return
 
         try:
@@ -696,9 +697,9 @@ def remove(self, itask, reason=""):
             self.main_pool_changed = True
             if not self.main_pool[itask.point]:
                 del self.main_pool[itask.point]
-                self.task_queue_mgr.remove_task(itask)
-                if itask.tdef.max_future_prereq_offset is not None:
-                    self.set_max_future_offset()
+            self.task_queue_mgr.remove_task(itask)
+            if itask.tdef.max_future_prereq_offset is not None:
+                self.set_max_future_offset()
 
             # Notify the data-store manager of their removal
             # (the manager uses window boundary tracking for pruning).
@@ -815,8 +816,8 @@ def release_queued_tasks(self):
 
         for itask in released:
             itask.state_reset(is_queued=False)
-            itask.waiting_on_job_prep = True
             self.data_store_mgr.delta_task_queued(itask)
+            itask.waiting_on_job_prep = True
 
             if cylc.flow.flags.cylc7_back_compat:
                 # Cylc 7 Back Compat: spawn downstream to cause Cylc 7 style
@@ -994,8 +995,7 @@ def can_stop(self, stop_mode):
                 and itask.state(*TASK_STATUSES_ACTIVE)
                 and not itask.state.kill_failed
             )
-            # we don't need to check for preparing tasks because they will be
-            # reset to waiting on restart
+            # preparing tasks because they will be reset to waiting on restart
             for itask in self.get_tasks()
         )
 
@@ -1321,7 +1321,7 @@ def remove_if_complete(self, itask):
                 )
             else:
                 # Remove as completed.
-                self.remove(itask, 'finished')
+                self.remove(itask, 'completed')
                 if itask.identity == self.stop_task_id:
                     self.stop_task_finished = True
                 if self.compute_runahead():
@@ -1701,21 +1701,24 @@ def _set_expired_task(self, itask):
                 or itask.tdef.expiration_offset is None
         ):
             return False
+
         if itask.expire_time is None:
             itask.expire_time = (
                 itask.get_point_as_seconds() +
                 itask.get_offset_as_seconds(itask.tdef.expiration_offset))
-        if time() > itask.expire_time:
-            msg = 'Task expired (skipping job).'
+
+        if (
+            time() > itask.expire_time and
+            itask.state_reset(TASK_STATUS_EXPIRED)
+        ):
+            msg = 'Task expired: will not submit job.'
             LOG.warning(f"[{itask}] {msg}")
             self.task_events_mgr.setup_event_handlers(itask, "expired", msg)
-            # TODO succeeded and expired states are useless due to immediate
-            # removal under all circumstances (unhandled failed is still used).
-            if itask.state_reset(TASK_STATUS_EXPIRED, is_held=False):
-                self.data_store_mgr.delta_task_state(itask)
-                self.data_store_mgr.delta_task_held(itask)
+            self.data_store_mgr.delta_task_state(itask)
+            self.spawn_on_output(itask, "expired")
             self.remove(itask, 'expired')
             return True
+
         return False
 
     def task_succeeded(self, id_):

tests/functional/hold-release/05-release.t

@@ -65,7 +65,7 @@ init_workflow "${TEST_NAME_BASE}" <<'__FLOW_CONFIG__'
         inherit = STOP
         script = """
             cylc__job__poll_grep_workflow_log -E \
-               '1/dog1 succeeded .* task proxy removed \(finished\)'
+               '1/dog1 succeeded .* task proxy removed \(completed\)'
             cylc stop "${CYLC_WORKFLOW_ID}"
         """
 __FLOW_CONFIG__

tests/functional/triggering/21-expire.t

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+# 
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#-------------------------------------------------------------------------------
+# Test expire triggering
+. "$(dirname "$0")/test_header"
+set_test_number 2
+reftest
+exit

tests/functional/triggering/21-expire/flow.cylc

@@ -0,0 +1,11 @@
+[scheduling]
+    initial cycle point = 1999
+    [[special tasks]]
+        clock-expire = foo(PT0S)
+    [[graph]]
+        R1 = """
+            foo:expire? => bar
+            foo => baz
+        """
+[runtime]
+   [[foo, bar, baz]]

tests/functional/triggering/21-expire/reference.log

@@ -0,0 +1 @@
+19990101T0000Z/bar -triggered off ['19990101T0000Z/foo'] in flow 1