ensure cancellation of task is not only in 1 process

Author: sanderegg

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

@@ -140,7 +140,6 @@ async def _triage_changed_tasks(
 class ScheduledPipelineParams:
     cluster_id: ClusterID
     run_metadata: RunMetadataDict
-    mark_for_cancellation: datetime.datetime | None
     use_on_demand_clusters: bool
 
     scheduler_task: asyncio.Task | None = None
@@ -198,7 +197,6 @@ async def run_new_pipeline(
             cluster_id=cluster_id,
             run_metadata=new_run.metadata,
             use_on_demand_clusters=use_on_demand_clusters,
-            mark_for_cancellation=None,
         )
         await publish_project_log(
             self.rabbitmq_client,
@@ -236,9 +234,6 @@ async def stop_pipeline(
         )
         if updated_comp_run:
             assert updated_comp_run.cancelled is not None  # nosec
-            self.scheduled_pipelines[
-                (user_id, project_id, selected_iteration)
-            ].mark_for_cancellation = updated_comp_run.cancelled
             # ensure the scheduler starts right away
             self.scheduled_pipelines[
                 (user_id, project_id, selected_iteration)
@@ -714,7 +709,10 @@ async def _schedule_pipeline(
                     project_id, dag
                 )
                 # 3. do we want to stop the pipeline now?
-                if pipeline_params.mark_for_cancellation:
+                comp_run = await CompRunsRepository.instance(self.db_engine).get(
+                    user_id, project_id, iteration
+                )
+                if comp_run.cancelled:
                     await self._schedule_tasks_to_stop(
                         user_id, project_id, comp_tasks, pipeline_params
                     )

services/director-v2/tests/unit/with_dbs/test_modules_comp_scheduler_dask_scheduler.py

@@ -402,7 +402,6 @@ async def test_misconfigured_pipeline_is_not_scheduled(
         assert u_id == user["id"]
         assert p_id == sleepers_project.uuid
         assert it > 0
-        assert params.mark_for_cancellation is None
     # check the database was properly updated
     async with aiopg_engine.acquire() as conn:
         result = await conn.execute(
@@ -450,7 +449,6 @@ async def _assert_start_pipeline(
         assert u_id == published_project.project.prj_owner
         assert p_id == published_project.project.uuid
         assert it > 0
-        assert params.mark_for_cancellation is None
         assert params.run_metadata == run_metadata
 
     # check the database is correctly updated, the run is published