Merge pull request #162 from openforcefield/bugfix/issue-138

Only action `Task`s that are `status` 'waiting', 'error', or 'running'

alchemiscale/interface/client.py

@@ -607,6 +607,7 @@ def action_tasks(
         A Task cannot be actioned:
             - to an AlchemicalNetwork in a different Scope.
             - if it extends another Task that is not complete.
+            - if it has any status other than 'waiting', 'running', or 'error'
 
         Parameters
         ----------

alchemiscale/storage/statestore.py

@@ -1074,6 +1074,9 @@ def action_tasks(
         A given compute task can be represented in any number of
         AlchemicalNetwork TaskHubs, or none at all.
 
+        Only Tasks with status 'waiting', 'running', or 'error' can be
+        actioned.
+
         """
         with self.transaction() as tx:
             actioned_sks = []
@@ -1086,8 +1089,10 @@ def action_tasks(
                 MATCH (task:Task {{_scoped_key: '{t}'}})-[:PERFORMS]->(tf:Transformation)<-[:DEPENDS_ON]-(an)
 
                 // only proceed for cases where task is not already actioned on hub
+                // and where the task is either in 'waiting', 'running', or 'error' status
                 WITH th, an, task
                 WHERE NOT (th)-[:ACTIONS]->(task)
+                  AND task.status IN ['waiting', 'running', 'error']
 
                 // create the connection
                 CREATE (th)-[ar:ACTIONS {{weight: 1.0}}]->(task)

alchemiscale/tests/integration/storage/test_statestore.py

@@ -822,6 +822,35 @@ def test_action_task(self, n4js: Neo4jStore, network_tyk2, scope_test):
         task_sks_fail = n4js.action_tasks(task_sks, taskhub_sk2)
         assert all([i is None for i in task_sks_fail])
 
+    def test_action_task_other_statuses(
+        self, n4js: Neo4jStore, network_tyk2, scope_test
+    ):
+        an = network_tyk2
+        network_sk = n4js.create_network(an, scope_test)
+        taskhub_sk: ScopedKey = n4js.create_taskhub(network_sk)
+
+        transformation = list(an.edges)[0]
+        transformation_sk = n4js.get_scoped_key(transformation, scope_test)
+
+        # create 10 tasks
+        task_sks = [n4js.create_task(transformation_sk) for i in range(6)]
+
+        # set all but first task to running
+        n4js.set_task_running(task_sks[1:])
+
+        # set 1 task for each available status
+        n4js.set_task_error(task_sks[2:3])
+        n4js.set_task_complete(task_sks[3:4])
+        n4js.set_task_invalid(task_sks[4:5])
+        n4js.set_task_deleted(task_sks[5:6])
+
+        # action all tasks; only those that are 'waiting', 'running', or
+        # 'error' should be actioned
+        actioned = n4js.action_tasks(task_sks, taskhub_sk)
+
+        assert actioned[:3] == task_sks[:3]
+        assert actioned[3:] == [None] * 3
+
     def test_action_task_extends(self, n4js: Neo4jStore, network_tyk2, scope_test):
         an = network_tyk2
         network_sk = n4js.create_network(an, scope_test)