Keep a finished workflow's node statuses current instead of expiring them

docs/whats_new.rst

@@ -42,6 +42,7 @@ See: :doc:`tethys_sdk/gizmos/time_picker`
 Bug Fixes
 ---------
 
+* Node statuses of finished condor workflows are served from the database instead of expiring: `PR 1298 <https://github.com/tethysplatform/tethys/pull/1298>`_
 * Login and register form fixes: `PR 1293 <https://github.com/tethysplatform/tethys/pull/1293>`_
 * Static file discovery fix for ``STATICFILES_USE_NPM``: `PR 1291 <https://github.com/tethysplatform/tethys/pull/1291>`_
 * Django 5 app initialization warning fix: `PR 1288 <https://github.com/tethysplatform/tethys/pull/1288>`_

tests/unit_tests/test_tethys_compute/test_models/test_CondorWorkflow.py

@@ -371,6 +371,49 @@ def test_node_statuses_max_age_is_overridable(self):
 
         self.assertFalse(self.condorworkflow.node_statuses_are_current)
 
+    def set_node_statuses(self, *statuses):
+        nodes = (self.condorworkflowjobnode, self.condorworkflowjobnode_child)
+        for node, status in zip(nodes, statuses):
+            node.cached_node_status = status
+            node.save()
+
+    def test_finished_workflow_stays_current_however_old(self):
+        self.condorworkflow._status = "COM"
+        self.set_node_statuses("Completed", "Completed")
+        self.condorworkflow.node_statuses_updated = tz.now() - datetime.timedelta(
+            days=30
+        )
+
+        self.assertTrue(self.condorworkflow.node_statuses_are_current)
+
+    def test_finished_workflow_with_a_node_left_running_still_expires(self):
+        self.condorworkflow._status = "COM"
+        self.set_node_statuses("Completed", "Running")
+        self.condorworkflow.node_statuses_updated = tz.now() - datetime.timedelta(
+            days=30
+        )
+
+        self.assertFalse(self.condorworkflow.node_statuses_are_current)
+
+    def test_unfinished_workflow_expires_even_with_every_node_terminal(self):
+        self.condorworkflow._status = "VAR"
+        self.set_node_statuses("Completed", "Completed")
+        self.condorworkflow.node_statuses_updated = tz.now() - datetime.timedelta(
+            days=30
+        )
+
+        self.assertFalse(self.condorworkflow.node_statuses_are_current)
+
+    def test_a_removed_node_counts_as_terminal(self):
+        self.set_node_statuses("Completed", "Removed")
+
+        self.assertTrue(self.condorworkflow.all_node_statuses_are_terminal)
+
+    def test_nodes_are_not_terminal_until_every_one_is_heard_from(self):
+        self.set_node_statuses("Completed")
+
+        self.assertFalse(self.condorworkflow.all_node_statuses_are_terminal)
+
     def test_cached_node_statuses_makes_no_remote_call(self):
         with mock.patch(
             "tethys_compute.models.condor.condor_workflow.CondorBase.condor_object"

tethys_compute/models/condor/condor_workflow.py

@@ -109,11 +109,47 @@ def node_statuses_are_current(self):
         longer per pass than the interval for one job, so tying the two together
         would leave the statuses never current precisely when there is enough load
         for it to matter.
+
+        Age stops mattering once there is nothing left to learn. A workflow that has
+        left the queue with every node in a status it cannot leave will never report
+        again, so expiring its statuses only sends the views to the scheduler to be
+        told the same thing -- and a finished workflow is the one whose working
+        directory is most likely to have been cleaned up, which is what reading a
+        node's live status needs.
         """
         if self.node_statuses_updated is None:
             return False
+        if (
+            self.cached_status in self.TERMINAL_STATUSES
+            and self.all_node_statuses_are_terminal
+        ):
+            return True
         return timezone.now() - self.node_statuses_updated < self.node_statuses_max_age
 
+    @property
+    def all_node_statuses_are_terminal(self):
+        """Whether every node has reached a status it cannot leave.
+
+        Both halves are load-bearing. The workflow's own status is not enough,
+        because a node can be reported terminal and then requeued -- a held node
+        that DAGMan retries -- and the workflow is only terminal once DAGMan itself
+        has departed. The nodes are not enough either: a status recorded by a
+        reporter that never sent a final one leaves the last node non-terminal, and
+        that is exactly the copy that must not be trusted forever.
+
+        A node with no persisted status has not been heard about at all, so it
+        counts against the whole set rather than being skipped.
+
+        Returns: bool
+        """
+        statuses = list(self.node_set.values_list("cached_node_status", flat=True))
+        if not statuses:
+            return False
+        return all(
+            self.STATUS_MAP.get(status) in self.TERMINAL_STATUS_CODES
+            for status in statuses
+        )
+
     @property
     def cached_node_statuses(self):
         """Node status counts built from the database instead of the scheduler.