Avoid spurious logging on restart with waiting parentless tasks.

Author: hjoliver

cylc/flow/task_pool.py

@@ -266,12 +266,22 @@ def db_add_new_flow_rows(self, itask: TaskProxy) -> None:
         self.workflow_db_mgr.put_insert_task_outputs(itask)
 
     def add_to_pool(self, itask) -> None:
-        """Add a task to the pool."""
+        """Add a task to the pool, if not already added."""
 
         self.active_tasks.setdefault(itask.point, {})
+        if itask.identity in self.active_tasks[itask.point]:
+            # This is normal on restart with waiting parentless tasks that
+            # auto-spawned to the runahead limit before shutdown (e.g. if
+            # you shut down immediately after starting up paused). Each
+            # will be resurrected from the DB again *and* try to auto-spawn
+            # again because we don't record that parentless tasks spawned.
+            # (But debug-log it anyway for any unexpected cases).
+            LOG.debug(f"{itask.identity} not added to n=0: already exists")
+            return None
+
         self.active_tasks[itask.point][itask.identity] = itask
         self.active_tasks_changed = True
-        LOG.debug(f"[{itask}] added to the n=0 window")
+        LOG.info(f"[{itask}] added to the n=0 window")
 
         self.create_data_store_elements(itask)
 
@@ -1835,13 +1845,20 @@ def spawn_task(
         if (
             prev_status is not None
             and not itask.state.outputs.get_completed_outputs()
+            and not self.config.experimental.expire_triggers
         ):
-            # If itask has any history in this flow but no completed outputs
-            # we can infer it has just been deliberately removed (N.B. not
-            # by `cylc remove`), so don't immediately respawn it.
-            # TODO (follow-up work):
-            # - this logic fails if task removed after some outputs completed
-            LOG.info(f"Not respawning {point}/{name} - task was removed")
+            # If itask has history but no completed outputs, it was removed
+            # by suicide trigger (not by `cylc remove` which erases history).
+            # This logic fails if suicided after completion of outputs!
+
+            # NOTE: redoing suicide triggers as expire triggers fixed this.
+            # TODO: remove this code once that's no longer experimental.
+            # Until then, this code also prevents double-spawning of waiting
+            # parentless tasks at restart if experimental expire triggers are
+            # off (but that is also handled properly without this block - by
+            # not adding tasks to the pool if already added (see comments in
+            # the add_to_pool method).
+            LOG.debug(f"Not respawning {point}/{name}")
             return None
 
         if prev_status in TASK_STATUSES_FINAL:

tests/integration/test_task_pool.py

@@ -1951,13 +1951,14 @@ async def test_fast_respawn(
     foo = task_pool.get_task(IntegerPoint("1"), "foo")
 
     # remove it from the pool
+    # (this is not the higher level "cylc remove" that would erase history)
     task_pool.remove(foo)
     assert foo not in task_pool.get_tasks()
 
     # attempt to spawn it again
     itask = task_pool.spawn_task("foo", IntegerPoint("1"), {1})
     assert itask is None
-    assert "Not respawning 1/foo - task was removed" in caplog.text
+    assert "Not respawning 1/foo" in caplog.text
 
 
 async def test_remove_active_task(
@@ -2576,3 +2577,31 @@ def list_spawn_task_calls():
         assert db_select(
             schd, True, 'task_outputs', 'outputs', cycle='1', name='foo'
         ) == [('{"x": "(manually completed)"}',)]
+
+
+async def test_add_to_pool(
+    flow, scheduler, start, caplog
+):
+    """It should log attempts to add the same task again."""
+    id_ = flow(
+        {
+            'scheduling': {
+                'graph': {
+                    'R1': 'a',
+                },
+            },
+        }
+    )
+    schd = scheduler(id_)
+
+    async with start(schd):
+        caplog.set_level(logging.DEBUG, CYLC_LOG)
+
+        # 1/a should be pre-spawned (parentless)
+        a_1 = schd.pool.get_task(IntegerPoint('1'), 'a')
+        assert a_1
+
+        # add it again
+        schd.pool.add_to_pool(a_1)
+
+        assert "1/a not added to n=0: already exists" in caplog.text