fix ordering of tasks

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_base.py

@@ -116,7 +116,7 @@ class SortedTasks:
 
 
 async def _triage_changed_tasks(
-    changed_tasks: list[tuple[_Previous, _Current]]
+    changed_tasks: list[tuple[_Previous, _Current]],
 ) -> SortedTasks:
     started_tasks = [
         current
@@ -242,9 +242,13 @@ async def _set_schedule_done(
     async def _set_states_following_failed_to_aborted(
         self, project_id: ProjectID, dag: nx.DiGraph
     ) -> dict[NodeIDStr, CompTaskAtDB]:
-        tasks: dict[NodeIDStr, CompTaskAtDB] = await self._get_pipeline_tasks(
-            project_id, dag
-        )
+        tasks = await self._get_pipeline_tasks(project_id, dag)
+        # Perform a reverse topological sort to ensure tasks are ordered from last to first
+        sorted_node_ids = list(reversed(list(nx.topological_sort(dag))))
+        tasks = {
+            node_id: tasks[node_id] for node_id in sorted_node_ids if node_id in tasks
+        }
+        # we need the tasks ordered from the last task to the first
         node_ids_to_set_as_aborted: set[NodeIDStr] = set()
         for task in tasks.values():
             if task.state == RunningState.FAILED:
@@ -651,8 +655,10 @@ async def _schedule_tasks_to_stop(
     ) -> None:
         # get any running task and stop them
         comp_tasks_repo = CompTasksRepository.instance(self.db_engine)
-        await comp_tasks_repo.mark_project_published_waiting_for_cluster_tasks_as_aborted(
-            project_id
+        await (
+            comp_tasks_repo.mark_project_published_waiting_for_cluster_tasks_as_aborted(
+                project_id
+            )
         )
         # stop any remaining running task, these are already submitted
         if tasks_to_stop := [

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_scheduler_dask.py

@@ -1451,7 +1451,6 @@ async def mocked_get_task_result(_job_id: str) -> TaskOutputData:
         project_uuid=running_project.project.uuid,
         task_ids=[
             running_project.tasks[1].node_id,
-            running_project.tasks[2].node_id,
             running_project.tasks[3].node_id,
         ],
         expected_state=reboot_state.expected_task_state_group1,
@@ -1460,7 +1459,7 @@ async def mocked_get_task_result(_job_id: str) -> TaskOutputData:
     await assert_comp_tasks(
         sqlalchemy_async_engine,
         project_uuid=running_project.project.uuid,
-        task_ids=[running_project.tasks[4].node_id],
+        task_ids=[running_project.tasks[2].node_id, running_project.tasks[4].node_id],
         expected_state=reboot_state.expected_task_state_group2,
         expected_progress=reboot_state.expected_task_progress_group2,
     )