Merge pull request #6856 from oliver-sanders/6821

suicide trigger should not clear previously completed outputs

Author: hjoliver

changes.d/6856.fix.md

@@ -0,0 +1 @@
+Fix a niche bug where outputs of a task could be wiped from the database if it was subsequently suicide triggered (e.g, if a custom output was manually set before the suicide trigger occurred).

cylc/flow/task_pool.py

@@ -1820,7 +1820,11 @@ def spawn_task(
                     for cycle, task, output in self.abs_outputs_done
                 ])
 
-        self.db_add_new_flow_rows(itask)
+        if prev_status is None:
+            # only add new flow rows if this task has not run before
+            # see https://github.com/cylc/cylc-flow/pull/6821
+            self.db_add_new_flow_rows(itask)
+
         return itask
 
     def _spawn_after_flow_wait(self, itask: TaskProxy) -> None:

tests/integration/test_task_pool.py

@@ -46,7 +46,7 @@
     TASK_OUTPUT_FAILED,
     TASK_OUTPUT_SUCCEEDED,
 )
-from cylc.flow.task_proxy import TaskProxy
+from cylc.flow.task_pool import TaskPool
 from cylc.flow.task_state import (
     TASK_STATUS_EXPIRED,
     TASK_STATUS_FAILED,
@@ -62,6 +62,7 @@
 if TYPE_CHECKING:
     from cylc.flow.cycling import PointBase
     from cylc.flow.scheduler import Scheduler
+    from cylc.flow.task_proxy import TaskProxy
 
 # NOTE: foo and bar have no parents so at start-up (even with the workflow
 # paused) they get spawned out to the runahead limit. 2/pub spawns
@@ -2438,3 +2439,75 @@ async def test_start_tasks(
                 "2050/baz",
             ])
         )
+
+
+async def test_add_new_flow_rows_on_spawn(
+    flow,
+    scheduler,
+    run,
+    complete,
+    db_select,
+    capcall,
+) -> None:
+    """Task suicide should not override previously completed outputs.
+
+    See https://github.com/cylc/cylc-flow/pull/6821
+    """
+    # capture all TaskPool.spawn_task() calls
+    spawn_task_calls = capcall(
+        'cylc.flow.task_pool.TaskPool.spawn_task', TaskPool.spawn_task
+    )
+
+    def list_spawn_task_calls():
+        """Return a list of the names of tasks which have been run through the
+        "spawn_tasks" function so far."""
+        return [
+            args[1] for args, _kwargs in spawn_task_calls
+        ]
+
+    id_ = flow({
+        'scheduling': {
+            'graph': {
+                'R1': '''
+                    slow:fail? => foo
+                    slow? => !foo
+                    foo:x => x
+                ''',
+            },
+        },
+        'runtime': {
+            'foo': {
+                'outputs': {'x': 'xxx'}
+            },
+        },
+    })
+
+    schd = scheduler(id_, paused_start=False)
+    async with run(schd):
+        # 1/slow should spawn on startup
+        assert list_spawn_task_calls() == ['slow']
+
+        # set foo:x
+        await commands.run_cmd(
+            commands.set_prereqs_and_outputs(
+                schd, ['1/foo'], ['1'], ['x'], None
+            )
+        )
+        # 1/foo:x should be recorded in the DB:
+        assert db_select(
+            schd, True, 'task_outputs', 'outputs', cycle='1', name='foo'
+        ) == [('{"x": "(manually completed)"}',)]
+        # and 1/x should spawn:
+        assert list_spawn_task_calls() == ['slow', 'x']
+
+        # run the workflow until completion
+        await complete(schd, timeout=5)
+
+        # 1/foo should spawn as a result of the suicide trigger
+        assert list_spawn_task_calls() == ['slow', 'x', 'foo']
+
+        # the manually completed output should not have been overwritten by the
+        # suicide trigger
+        assert db_select(
+            schd, True, 'task_outputs', 'outputs', cycle='1', name='foo'
+        ) == [('{"x": "(manually completed)"}',)]