Fix cylc set on a task no longer in pool causing task state regression

Author: MetRonnie

cylc/flow/task_pool.py

@@ -2118,8 +2118,7 @@ def set_prereqs_and_outputs(
                         point, tdef, flow_nums,
                         flow_wait=flow_wait, transient=True
                     )
-                    if trans is not None:
-                        self._set_outputs_itask(trans, outputs)
+                    if trans and self._set_outputs_itask(trans, outputs):
                         no_op = False
 
         if not no_op:
@@ -2189,15 +2188,18 @@ def _set_outputs_itask(
         self,
         itask: 'TaskProxy',
         outputs: Iterable[str],
-    ) -> None:
+    ) -> bool:
         """Set requested outputs on a task proxy and spawn children.
 
         If no outputs were specified and the task has no required outputs to
         set, set the "success pathway" outputs in the same way that skip mode
         does.
 
         Designated flows should already be merged to the task proxy.
+
+        Returns True if any outputs were set, else False.
         """
+        no_op = True
         outputs = set(outputs)
 
         if not outputs:
@@ -2227,18 +2229,24 @@ def _set_outputs_itask(
                 LOG.info(f"output {itask.identity}:{output} completed already")
                 continue
             self.task_events_mgr.process_message(
-                itask, logging.INFO, output, forced=True)
+                itask, logging.INFO, output, forced=True
+            )
+            no_op = False
 
         if not itask.state(TASK_STATUS_WAITING):
             # Can't be runahead limited or queued.
             itask.state_reset(is_runahead=False, is_queued=False)
             self.task_queue_mgr.remove_task(itask)
 
+        if no_op:
+            return False
+
         self.data_store_mgr.delta_task_state(itask)
         self.data_store_mgr.delta_task_outputs(itask)
         self.workflow_db_mgr.put_update_task_state(itask)
         self.workflow_db_mgr.put_update_task_outputs(itask)
         self.workflow_db_mgr.process_queued_ops()
+        return True
 
     def _set_prereqs_itask(
         self,

tests/integration/scripts/test_set.py

@@ -369,3 +369,46 @@ def assert_job_failed(itask: TaskProxy):
         assert db_task_states(foo) == [(TASK_STATUS_SUCCEEDED,)]
         # But job is still failed:
         assert_job_failed(foo)
+
+
+async def test_set_already_succeeded(
+    flow, scheduler, run, complete, db_select
+):
+    """Doing `cylc set` on a task that has already succeeded should not
+    change anything."""
+    schd: Scheduler = scheduler(
+        flow('foo => bar'),
+        paused_start=False,
+    )
+
+    def db_task_states(itask: TaskProxy):
+        return db_select(
+            schd,
+            True,
+            'task_states',
+            'submit_num',
+            'status',
+            'time_updated',
+            name=itask.tdef.name,
+        )
+
+    def data_store_task_state(itask: TaskProxy):
+        return schd.data_store_mgr.data[schd.tokens.id][TASK_PROXIES][
+            itask.tokens.id
+        ].state
+
+    async with run(schd):
+        foo = schd.pool.get_tasks()[0]
+        await complete(schd, foo.identity)
+        time_updated = db_task_states(foo)[0][2]
+        expected = [(1, TASK_STATUS_SUCCEEDED, time_updated)]
+        assert db_task_states(foo) == expected
+        assert data_store_task_state(foo) == TASK_STATUS_SUCCEEDED
+
+        await run_cmd(
+            set_prereqs_and_outputs(schd, [foo.identity], [])
+        )
+        assert foo.state(TASK_STATUS_SUCCEEDED)
+        await schd.update_data_structure()
+        assert db_task_states(foo) == expected
+        assert data_store_task_state(foo) == TASK_STATUS_SUCCEEDED