restart timeout: fire event for stop after cycle point scenarios

* If a workflow hits the `final cycle point` and is restarted, it is
  considered complete.
* If a `restart timeout` is configured, the scheduler will stay up for a
  configured period.
* However, if a workflow hits the `stop after after cycle point` and is
  restarted, the `restart timeout` event is not yielded and the
  scheduler will instantly shut down again.

Author: oliver-sanders

changes.d/6903.fix.md

@@ -0,0 +1 @@
+Workflows that have hit the `stop after cycle point` will no longer shut down immediately when restart according to the `restart timeout` configuration.

cylc/flow/scheduler.py

@@ -555,8 +555,22 @@ async def configure(self, params):
                     timer.reset()
                 self.timers[event] = timer
 
-        if self.is_restart and not self.pool.get_tasks():
-            # This workflow completed before restart; wait for intervention.
+        if self.is_restart and (
+            # workflow has completed
+            not self.pool.get_tasks()
+            # workflow has hit the "stop after cycle point"
+            or (
+                self.config.stop_point
+                and all(
+                    cycle > self.config.stop_point
+                    for cycle in {
+                        itask.point for itask in self.pool.get_tasks()
+                    }
+                )
+            )
+        ):
+            # This workflow will shut down immediately once restarted
+            # => Give the user a grace period to intervene first
             with suppress(KeyError):
                 self.timers[self.EVENT_RESTART_TIMEOUT].reset()
                 self.is_restart_timeout_wait = True

tests/integration/test_workflow_events.py

@@ -51,20 +51,23 @@ async def test_scheduler(flow, scheduler, capcall):
     def get_events():
         return {e[0][1] for e in events}
 
-    def _schd(config=None, **opts):
-        id_ = flow({
-            'scheduler': {
-                'events': {
-                    'mail events': ', '.join(EVENTS),
-                    **(config or {}),
+    def _schd(event_config=None, config=None, **opts):
+        assert not (event_config and config)
+        if not config:
+            config = {
+                'scheduler': {
+                    'events': {
+                        'mail events': ', '.join(EVENTS),
+                        **(event_config or {}),
+                    },
                 },
-            },
-            'scheduling': {
-                'graph': {
-                    'R1': 'a'
-                }
-            },
-        })
+                'scheduling': {
+                    'graph': {
+                        'R1': 'a'
+                    }
+                },
+            }
+        id_ = flow(config)
         schd = scheduler(id_, **opts)
         schd.get_events = get_events
         return schd
@@ -168,8 +171,13 @@ async def test_stall(test_scheduler, start):
     assert schd.get_events() == {'shutdown', 'stall'}
 
 
-async def test_restart_timeout(test_scheduler, scheduler, run, complete):
-    """Test restart timeout.
+async def test_restart_timeout_workflow_completion(
+    test_scheduler,
+    scheduler,
+    run,
+    complete,
+):
+    """Test restart timeout for completed workflows.
 
     This should fire when a completed workflow is restarted.
     """
@@ -188,6 +196,46 @@ async def test_restart_timeout(test_scheduler, scheduler, run, complete):
     assert schd2.get_events() == {'startup', 'restart timeout', 'shutdown'}
 
 
+async def test_restart_timeout_workflow_stop_after_cycle_point(
+    test_scheduler,
+    scheduler,
+    run,
+    complete,
+):
+    """Test restart timeout with the "stop after cycle point" config.
+
+    This should fire when a completed workflow is restarted.
+    """
+    schd = test_scheduler(
+        config={
+            'scheduler': {
+                'cycle point format': 'CCYY',
+                'events': {'restart timeout': 'PT0S'},
+            },
+            'scheduling': {
+                'initial cycle point': '2000',
+                'stop after cycle point': '2000',
+                'graph': {
+                    'P1Y': 'foo[-P1Y] => foo',
+                },
+            },
+        },
+        paused_start=False,
+    )
+
+    # run to completion
+    async with run(schd):
+        await complete(schd)
+    assert schd.get_events() == {'startup', 'shutdown'}
+
+    # restart
+    schd2 = scheduler(schd.workflow)
+    schd2.get_events = schd.get_events
+    async with run(schd2):
+        await asyncio.sleep(0.1)
+    assert schd2.get_events() == {'startup', 'restart timeout', 'shutdown'}
+
+
 async def test_shutdown_handler_timeout_kill(
     test_scheduler, run, monkeypatch, mock_glbl_cfg, caplog
 ):