Deprecate trigger --on-resume option. (#6974)

Remove obsolete trigger --on-resume option.

Author: hjoliver

changes.d/6974.feat.md

@@ -0,0 +1 @@
+Removed obsolete trigger --on-resume option.

cylc/flow/commands.py

@@ -652,6 +652,10 @@ async def force_trigger_tasks(
     flow: List[str],
     flow_wait: bool = False,
     flow_descr: Optional[str] = None,
+    # BACK COMPAT: on_resume
+    #   Arg no longer used but retained for older clients.
+    # From: 8.6
+    # Remove at: 8.7
     on_resume: bool = False
 ):
     """Match and trigger a group of tasks (`cylc trigger` command).
@@ -667,12 +671,6 @@ async def force_trigger_tasks(
     flow = back_compat_flow_all(flow)  # BACK COMPAT (see func def)
     ids = validate.is_tasks(tasks)
     validate.flow_opts(flow, flow_wait)
-    if on_resume:
-        LOG.warning(
-            "The --on-resume option is deprecated and will be removed "
-            "at Cylc 8.6."
-        )
-
     yield
 
     matched, unmatched = schd.pool.id_match(ids)
@@ -705,7 +703,6 @@ async def force_trigger_tasks(
             flow,
             flow_wait,
             flow_descr,
-            on_resume,
         )
 
 
@@ -715,7 +712,6 @@ def _force_trigger_tasks(
     flow: List[str],
     flow_wait: bool = False,
     flow_descr: Optional[str] = None,
-    on_resume: bool = False
 ):
     active = schd.pool.get_itasks(group_ids)
     if flow or not active:
@@ -792,7 +788,7 @@ def _force_trigger_tasks(
                 schd.pool.merge_flows(itask, flow_nums)
 
                 # Trigger group start task.
-                schd.pool.queue_or_trigger(itask, on_resume)
+                schd.pool.queue_or_trigger(itask)
 
         else:
             active_to_remove.append(itask.tokens.task)
@@ -883,6 +879,6 @@ def _force_trigger_tasks(
 
         if jtask is not None and not in_flow_prereqs:
             # Trigger group start task.
-            schd.pool.queue_or_trigger(jtask, on_resume)
+            schd.pool.queue_or_trigger(jtask)
 
     schd.pool.release_runahead_tasks()

cylc/flow/network/schema.py

@@ -2219,12 +2219,16 @@ class Meta:
         ''')
         resolver = partial(mutator, command='force_trigger_tasks')
 
+    # BACK COMPAT: on_resume
+    #   Arg no longer used but retained for older clients.
+    # From: 8.6
+    # Remove at: 8.7
     class Arguments(TaskMutation.Arguments, FlowMutationArguments):
         on_resume = Boolean(
             default_value=False,
             description=sstrip('''
-                If the workflow is paused, wait until it is resumed before
-                running the triggered task(s).
+                DEPRECATED: this option is no longer needed and will be
+                ignored by the scheduler.
             ''')
         )
 

cylc/flow/scheduler.py

@@ -683,9 +683,6 @@ async def run_scheduler(self) -> None:
 
                     # If we shut down with manually triggered waiting tasks,
                     # submit them to run now.
-                    # NOTE: this will run tasks that were triggered with
-                    # the trigger "--on-resume" option, even if the workflow
-                    # is restarted as paused. Option to be removed at 8.6.0.
                     pre_prep_tasks = []
                     for itask in self.pool.get_tasks():
                         if (
@@ -1349,7 +1346,7 @@ def release_tasks_to_run(self) -> bool:
             and self.reload_pending is False
         ):
             if self.pool.tasks_to_trigger_now:
-                # manually triggered tasks to run now, workflow paused or not
+                # manually triggered tasks to run now.
                 pre_prep_tasks.update(self.pool.tasks_to_trigger_now)
                 self.pool.tasks_to_trigger_now = set()
 
@@ -1362,10 +1359,6 @@ def release_tasks_to_run(self) -> bool:
             else:
                 # release queued tasks
                 pre_prep_tasks.update(self.pool.release_queued_tasks())
-                if self.pool.tasks_to_trigger_on_resume:
-                    # and manually triggered tasks to run once workflow resumed
-                    pre_prep_tasks.update(self.pool.tasks_to_trigger_on_resume)
-                    self.pool.tasks_to_trigger_on_resume = set()
 
         elif (
             (

cylc/flow/scripts/trigger.py

@@ -76,15 +76,13 @@
   $flow: [Flow!],
   $flowWait: Boolean,
   $flowDescr: String,
-  $onResume: Boolean,
 ) {
   trigger (
     workflows: $wFlows,
     tasks: $tasks,
     flow: $flow,
     flowWait: $flowWait,
     flowDescr: $flowDescr,
-    onResume: $onResume,
   ) {
     result
   }
@@ -103,17 +101,6 @@ def get_option_parser() -> COP:
 
     add_flow_opts_for_trigger_and_set(parser)
 
-    parser.add_option(
-        "--on-resume",
-        help=(
-            "If the workflow is paused, wait until it is resumed before "
-            "running the triggered task(s). DEPRECATED - this will be "
-            "removed at Cylc 8.6."
-        ),
-        action="store_true",
-        default=False,
-        dest="on_resume"
-    )
     return parser
 
 
@@ -131,7 +118,6 @@ async def run(options: 'Values', workflow_id: str, *tokens_list):
             'flow': options.flow,
             'flowWait': options.flow_wait,
             'flowDescr': options.flow_descr,
-            'onResume': options.on_resume,
         }
     }
     return await pclient.async_request('graphql', mutation_kwargs)

cylc/flow/task_pool.py

@@ -212,7 +212,6 @@ def __init__(
 
         self.tasks_to_hold: Set[Tuple[str, 'PointBase']] = set()
         self.tasks_to_trigger_now: Set['TaskProxy'] = set()
-        self.tasks_to_trigger_on_resume: Set['TaskProxy'] = set()
 
     def set_stop_task(self, task_id):
         """Set stop after a task."""
@@ -904,8 +903,6 @@ def remove(self, itask: 'TaskProxy', reason: Optional[str] = None) -> None:
         else:
             with suppress(KeyError):
                 self.tasks_to_trigger_now.remove(itask)
-            with suppress(KeyError):
-                self.tasks_to_trigger_on_resume.remove(itask)
             self.tasks_removed = True
             self.active_tasks_changed = True
             if not self.active_tasks[itask.point]:
@@ -2339,7 +2336,7 @@ def _get_active_flow_nums(self) -> 'FlowNums':
             or {1}
         )
 
-    def queue_or_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
+    def queue_or_trigger(self, itask: 'TaskProxy'):
         """Handle state, queues, and runahead for a manually triggered task.
 
         Triggering a non-queued task:
@@ -2349,7 +2346,7 @@ def queue_or_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
         Triggering a queued task:
           - run it, regardless of the queue limit
 
-        If ready, add itask to the tasks_to_trigger_(now/on_resume) lists.
+        If ready, add itask to the tasks_to_trigger_now list.
 
         Assumes the task is in the pool.
 
@@ -2389,16 +2386,7 @@ def queue_or_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
         if not itask.state.is_queued:
             # If not queued now, record the task as ready to run.
             itask.waiting_on_job_prep = True
-
-            if on_resume:
-                self.tasks_to_trigger_on_resume.add(itask)
-                # In case previously triggered without --on-resume.
-                # (It should have run already, but just in case).
-                self.tasks_to_trigger_now.discard(itask)
-            else:
-                self.tasks_to_trigger_now.add(itask)
-                # In case previously triggered with --on-resume.
-                self.tasks_to_trigger_on_resume.discard(itask)
+            self.tasks_to_trigger_now.add(itask)
 
         # Task may be set running before xtrigger is satisfied,
         # if so check/spawn if xtrigger sequential.

tests/integration/test_force_trigger.py

@@ -152,69 +152,6 @@ async def test_trigger_group_whilst_paused(flow, scheduler, run, complete):
         await complete(schd, '1/c')
 
 
-async def test_trigger_on_resume(
-    flow: 'Fixture',
-    scheduler: 'Fixture',
-    start: 'Fixture',
-    capture_submission: 'Fixture',
-):
-    """
-    Test manual triggering on-resume option when the workflow is paused.
-
-    https://github.com/cylc/cylc-flow/issues/6192
-
-    """
-    id_ = flow({
-        'scheduling': {
-            'queues': {
-                'default': {
-                    'limit': 1,
-                },
-            },
-            'graph': {
-                'R1': '''
-                    a => x & y & z
-                ''',
-            },
-        },
-    })
-    schd = scheduler(id_, paused_start=True)
-
-    # start the scheduler (but don't set the main loop running)
-    async with start(schd):
-
-        # capture task submissions (prevents real submissions)
-        submitted_tasks = capture_submission(schd)
-
-        # paused at start-up so no tasks should be submitted
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/x - it not should be submitted
-        await run_cmd(
-            force_trigger_tasks(schd, ['1/x'], ["1"], on_resume=True))
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/y - it should not be submitted
-        # (queue limit reached)
-        await run_cmd(
-            force_trigger_tasks(schd, ['1/y'], ["1"], on_resume=True))
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/y again - it should not be submitted
-        # (triggering a queued task runs it)
-        await run_cmd(
-            force_trigger_tasks(schd, ['1/y'], ["1"], on_resume=True))
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # resume the workflow, both tasks should trigger now.
-        schd.resume_workflow()
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 2
-
-
 async def test_trigger_group(
     flow, scheduler, run, complete, log_filter
 ):

tests/integration/test_remove.py

@@ -519,7 +519,7 @@ async def test_reload_changed_config(flow, scheduler, run, complete):
 
 
 async def test_remove_triggered(flow, scheduler, start):
-    """It should remove tasks from pool and from to_trigger sets."""
+    """It should remove tasks from pool and from to_trigger set."""
     conf = {
         'scheduling': {
             'graph': {
@@ -536,15 +536,14 @@ async def test_remove_triggered(flow, scheduler, start):
         )
         assert foo in schd.pool.tasks_to_trigger_now
 
-        # trigger bar (on resume)
+        # trigger bar
         await run_cmd(
-            force_trigger_tasks(schd, [bar.identity], [], on_resume=True)
+            force_trigger_tasks(schd, [bar.identity], [])
         )
-        assert bar in schd.pool.tasks_to_trigger_on_resume
+        assert bar in schd.pool.tasks_to_trigger_now
 
         await run_cmd(
             remove_tasks(schd, [foo.identity, bar.identity], [])
         )
         assert not schd.pool.get_tasks()
         assert not schd.pool.tasks_to_trigger_now
-        assert not schd.pool.tasks_to_trigger_on_resume