expire: fix interaction with automatic retries

* Closes #6717

Author: oliver-sanders

changes.d/7056.fix.md

@@ -0,0 +1 @@
+Fixes the interaction between retrying tasks and the expired state.

cylc/flow/etc/examples/expiry/.validate

@@ -30,7 +30,7 @@ test_one () {
         ./one
 
     # the start task should have expired
-    grep 'start.*(internal)expired' "$HOME/cylc-run/$ID/log/scheduler/log"
+    grep 'start.*expired' "$HOME/cylc-run/$ID/log/scheduler/log"
 
     # the following task(s) should not have run
     grep 'a.*running' "$HOME/cylc-run/$ID/log/scheduler/log" && exit 1
@@ -60,7 +60,7 @@ test_two () {
     grep 'start.*running' "$HOME/cylc-run/$ID/log/scheduler/log"
 
     # some other task in the chain should expire
-    grep '(internal)expired' "$HOME/cylc-run/$ID/log/scheduler/log"
+    grep 'expired' "$HOME/cylc-run/$ID/log/scheduler/log"
 
     # the housekeep task at the end of the cycle should not run
     grep 'housekeep.*running' "$HOME/cylc-run/$ID/log/scheduler/log" && exit 1
@@ -86,7 +86,7 @@ test_three () {
         ./three
 
     # the start task should expire
-    grep 'start.*(internal)expired' "$HOME/cylc-run/$ID/log/scheduler/log"
+    grep 'start.*expired' "$HOME/cylc-run/$ID/log/scheduler/log"
     # shellcheck disable=SC2125 # could only ever be one matching file
     local job_file="$HOME/cylc-run/$ID/log/job/"*"/a/NN/job"
     [[ ! -f "$job_file" ]]

cylc/flow/task_events_mgr.py

@@ -718,7 +718,8 @@ def process_message(
                 The submit number of the task relevant for the message.
                 If not specified, use latest submit number.
             forced:
-                If this message is due to manual completion or not (cylc set)
+                True, if this message is due to manual (as opposed to natural)
+                output completion.
 
         Return:
             False: in normal circumstances.

cylc/flow/task_pool.py

@@ -1581,7 +1581,7 @@ def spawn_on_output(self, itask: TaskProxy, output: str) -> None:
             if self.config.experimental.expire_triggers:
                 self.task_queue_mgr.remove_task(c_task)
                 self.task_events_mgr.process_message(
-                    c_task, logging.WARNING, TASK_OUTPUT_EXPIRED
+                    c_task, logging.WARNING, TASK_OUTPUT_EXPIRED, forced=True
                 )
             else:
                 self.remove(c_task, self.__class__.SUICIDE_MSG)
@@ -2476,6 +2476,7 @@ def clock_expire_tasks(self):
                     itask,
                     logging.WARNING,
                     TASK_OUTPUT_EXPIRED,
+                    forced=True,
                 )
 
     def task_succeeded(self, id_):

tests/integration/test_task_pool.py

@@ -42,6 +42,7 @@
 from cylc.flow.exceptions import WorkflowConfigError
 from cylc.flow.flow_mgr import FLOW_NONE
 from cylc.flow.id import TaskTokens, Tokens
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_events_mgr import TaskEventsManager
 from cylc.flow.task_outputs import (
     TASK_OUTPUT_FAILED,
@@ -2289,7 +2290,9 @@ async def list_data_store():
         }
 
 
-@pytest.mark.parametrize('expire_type', ['clock-expire', 'manual'])
+@pytest.mark.parametrize(
+    'expire_type', ['clock-expire', 'manual-expire', 'expire-trigger']
+)
 async def test_expire_dequeue_with_retries(
     flow, scheduler, start, expire_type
 ):
@@ -2298,48 +2301,91 @@ async def test_expire_dequeue_with_retries(
     See https://github.com/cylc/cylc-flow/issues/6284
     """
     conf = {
+        'scheduler': {
+            'experimental': {
+                'expire triggers': True,
+            },
+        },
         'scheduling': {
             'initial cycle point': '2000',
-
             'graph': {
                 'R1': 'foo'
             },
         },
         'runtime': {
             'foo': {
-                'execution retry delays': 'PT0S'
-            }
-        }
+                'execution retry delays': 'PT0S',
+                'outputs': {
+                    'x': 'xxx',
+                },
+            },
+        },
     }
 
     if expire_type == 'clock-expire':
+        # configre foo to clock-expire
         conf['scheduling']['special tasks'] = {'clock-expire': 'foo(PT0S)'}
-        method = lambda schd: schd.pool.clock_expire_tasks()
+
+        # run the clock-expire logic
+        def method(schd):
+            schd.pool.clock_expire_tasks()
+
+    elif expire_type == 'manual-expire':
+        # run the "cylc set" command to expire "foo"
+        def method(schd):
+            schd.pool.set_prereqs_and_outputs(
+                {TaskTokens('20000101T0000Z', 'foo')},
+                prereqs=[],
+                outputs=['expired'],
+                flow=['1'],
+            )
+
     else:
-        method = lambda schd: schd.pool.set_prereqs_and_outputs(
-            {TaskTokens('20000101T0000Z', 'foo')},
-            prereqs=[],
-            outputs=['expired'],
-            flow=['1'],
-        )
+        # configure the task to be expire-triggered once "bar" succeeds
+        conf['scheduling']['graph']['R1'] = '''
+            foo:x => bar
+            foo:x & bar => !foo
+        '''
+
+        # run the "cylc set" command to succeed "bar"
+        def method(schd):
+            schd.pool.set_prereqs_and_outputs(
+                {TaskTokens('20000101T0000Z', 'bar')},
+                prereqs=[],
+                outputs=['succeeded'],
+                flow=['1'],
+            )
 
     id_ = flow(conf)
-    schd = scheduler(id_)
+    schd = scheduler(id_, run_mode='live')
     schd: Scheduler
     async with start(schd):
-        itask = schd.pool.get_tasks()[0]
+        foo = schd.pool._get_task_by_id('20000101T0000Z/foo')
+        assert foo
+
+        # fake a real submission failure
+        # NOTE: yes, all of these things are needed for a valid test!
+        # Try removing the "force=True" added in this commit and ensure the
+        # "clock-expire" test fails before changing anything here!
+        foo.submit_num += 1
+        foo.run_mode = RunMode.LIVE
+        schd.task_job_mgr._set_retry_timers(foo, foo.tdef.rtconfig)
+        schd.task_events_mgr.process_message(foo, 0, 'started')
+        schd.task_events_mgr.process_message(foo, 0, 'xxx')
+        schd.task_events_mgr.process_message(foo, 0, 'failed')
+        schd.task_events_mgr._retry_task(foo, 0)
 
         # the task should start as "waiting(queued)"
-        assert itask.state(TASK_STATUS_WAITING, is_queued=True)
+        assert foo.state(TASK_STATUS_WAITING, is_queued=True)
 
         # expire the task via whichever method we are testing
         method(schd)
 
         # the task should enter the "expired" state
-        assert itask.state(TASK_STATUS_EXPIRED, is_queued=False)
+        assert foo.state(TASK_STATUS_EXPIRED, is_queued=False)
 
         # the task should also have been removed from the queue
-        assert not schd.pool.task_queue_mgr.remove_task(itask)
+        assert not schd.pool.task_queue_mgr.remove_task(foo)
 
 
 async def test_clock_expire_with_sequential_xtriggers(