Fix data store job accumulation bug (#6656)

Author: MetRonnie

changes.d/6656.fix.md

@@ -0,0 +1 @@
+Fix bug where old cycle points could accumulate in the UI.

cylc/flow/data_store_mgr.py

@@ -1734,7 +1734,7 @@ def update_data_structure(self):
             self.update_workflow()
 
             # Don't process updated deltas of pruned nodes
-            self.prune_pruned_updated_nodes()
+            self.dedupe_pruned_updated_task_proxies()
 
             # Gather deltas
             self.batch_deltas()
@@ -1958,8 +1958,8 @@ def _family_ascent_point_prune(
         if fp_id in parent_ids:
             parent_ids.remove(fp_id)
 
-    def prune_pruned_updated_nodes(self):
-        """Remove updated nodes that will also be pruned this batch.
+    def dedupe_pruned_updated_task_proxies(self):
+        """Remove updated task proxies that will also be pruned in this batch.
 
         This will avoid processing and sending deltas that will immediately
         be pruned. Kept separate from other pruning to allow for update
@@ -1969,19 +1969,19 @@ def prune_pruned_updated_nodes(self):
         tp_data = self.data[self.workflow_id][TASK_PROXIES]
         tp_added = self.added[TASK_PROXIES]
         tp_updated = self.updated[TASK_PROXIES]
-        j_updated = self.updated[JOBS]
         for tp_id in self.pruned_task_proxies:
             if tp_id in tp_updated:
                 if tp_id in tp_data:
-                    node = tp_data[tp_id]
+                    node: PbTaskProxy = tp_data[tp_id]
                 elif tp_id in tp_added:
                     node = tp_added[tp_id]
                 else:
                     continue
-                update_node = tp_updated.pop(tp_id)
-                for j_id in list(node.jobs) + list(update_node.jobs):
-                    if j_id in j_updated:
-                        del j_updated[j_id]
+                update_node: PbTaskProxy = tp_updated.pop(tp_id)
+                # Remove this task's added/updated jobs in this batch too:
+                for j_id in set(node.jobs).union(update_node.jobs):
+                    for record in (self.added, self.updated):
+                        record[JOBS].pop(j_id, None)
                 self.n_window_edges.difference_update(update_node.edges)
                 self.deltas[EDGES].pruned.extend(update_node.edges)
         self.pruned_task_proxies.clear()

tests/integration/test_data_store_mgr.py

@@ -14,11 +14,19 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import Iterable, List, TYPE_CHECKING, cast
+from logging import INFO
+from typing import (
+    Iterable,
+    List,
+    cast,
+)
 
 import pytest
 
-from cylc.flow.data_messages_pb2 import PbPrerequisite, PbTaskProxy
+from cylc.flow.data_messages_pb2 import (
+    PbPrerequisite,
+    PbTaskProxy,
+)
 from cylc.flow.data_store_mgr import (
     EDGES,
     FAMILY_PROXIES,
@@ -31,22 +39,19 @@
 from cylc.flow.scheduler import Scheduler
 from cylc.flow.task_events_mgr import TaskEventsManager
 from cylc.flow.task_outputs import (
-    TASK_OUTPUT_SUBMITTED,
     TASK_OUTPUT_STARTED,
+    TASK_OUTPUT_SUBMITTED,
     TASK_OUTPUT_SUCCEEDED,
 )
 from cylc.flow.task_state import (
     TASK_STATUS_FAILED,
+    TASK_STATUS_PREPARING,
     TASK_STATUS_SUCCEEDED,
     TASK_STATUS_WAITING,
 )
 from cylc.flow.wallclock import get_current_time_string
 
 
-if TYPE_CHECKING:
-    from cylc.flow.scheduler import Scheduler
-
-
 # NOTE: Some of these tests mutate the data store, so running them in
 # isolation may see failures when they actually pass if you run the
 # whole file
@@ -503,3 +508,20 @@ def _patch_remove(*args, **kwargs):
             TASK_OUTPUT_SUCCEEDED,
         }
         assert get_delta_outputs() is None
+
+
+async def test_remove_added_jobs_of_pruned_task(one: Scheduler, start):
+    """When a task is pruned, any of its jobs added in the same batch
+    must be removed from the batch.
+
+    See https://github.com/cylc/cylc-flow/pull/6656
+    """
+    async with start(one):
+        itask = one.pool.get_tasks()[0]
+        itask.state_reset(TASK_STATUS_PREPARING)
+        one.task_events_mgr.process_message(itask, INFO, TASK_OUTPUT_SUCCEEDED)
+        assert not one.data_store_mgr.data[one.id][JOBS]
+        assert len(one.data_store_mgr.added[JOBS]) == 1
+        one.data_store_mgr.update_data_structure()
+        assert not one.data_store_mgr.data[one.id][JOBS]
+        assert not one.data_store_mgr.added[JOBS]