fix

Author: matusdrobuliak66

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

@@ -284,6 +284,7 @@ async def _send_running_tasks_heartbeat(
         self,
         user_id: UserID,
         project_id: ProjectID,
+        run_id: PositiveInt,
         iteration: Iteration,
         dag: nx.DiGraph,
     ) -> None:
@@ -322,7 +323,7 @@ def _need_heartbeat(task: CompTaskAtDB) -> bool:
             await asyncio.gather(
                 *(
                     comp_tasks_repo.update_project_task_last_heartbeat(
-                        t.project_id, t.node_id, utc_now
+                        t.project_id, t.node_id, run_id, utc_now
                     )
                     for t in running_tasks
                 )
@@ -643,7 +644,7 @@ async def apply(
                 )
                 # 5. send a heartbeat
                 await self._send_running_tasks_heartbeat(
-                    user_id, project_id, iteration, dag
+                    user_id, project_id, comp_run.run_id, iteration, dag
                 )
 
                 # 6. Update the run result

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

@@ -391,10 +391,10 @@ async def _task_progress_change_handler(
             node_id = task_progress_event.task_owner.node_id
             comp_tasks_repo = CompTasksRepository(self.db_engine)
             task = await comp_tasks_repo.get_task(project_id, node_id)
+            run = await CompRunsRepository(self.db_engine).get(user_id, project_id)
             if task.state in WAITING_FOR_START_STATES:
                 task.state = RunningState.STARTED
                 task.progress = task_progress_event.progress
-                run = await CompRunsRepository(self.db_engine).get(user_id, project_id)
                 await self._process_started_tasks(
                     [task],
                     user_id=user_id,
@@ -405,7 +405,7 @@ async def _task_progress_change_handler(
                 )
             else:
                 await comp_tasks_repo.update_project_task_progress(
-                    project_id, node_id, task_progress_event.progress
+                    project_id, node_id, run.run_id, task_progress_event.progress
                 )
             await publish_service_progress(
                 self.rabbitmq_client,

services/director-v2/src/simcore_service_director_v2/modules/dask_client.py

@@ -57,7 +57,7 @@
 from models_library.resource_tracker import HardwareInfo
 from models_library.services import ServiceRunID
 from models_library.users import UserID
-from pydantic import PositiveInt, TypeAdapter, ValidationError
+from pydantic import TypeAdapter, ValidationError
 from pydantic.networks import AnyUrl
 from servicelib.logging_utils import log_catch, log_context
 from settings_library.s3 import S3Settings
@@ -267,7 +267,6 @@ async def send_computation_tasks(
         *,
         user_id: UserID,
         project_id: ProjectID,
-        run_id: PositiveInt,
         tasks: dict[NodeID, Image],
         callback: _UserCallbackInSepThread,
         remote_fct: ContainerRemoteFct | None = None,

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

@@ -156,6 +156,7 @@ async def test_schedule_all_pipelines(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
     # this directly schedule a new pipeline
     scheduler_rabbit_client_parser.assert_called_once_with(
@@ -257,6 +258,7 @@ async def test_schedule_all_pipelines_logs_error_if_it_find_old_pipelines(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
     # this directly schedule a new pipeline
     scheduler_rabbit_client_parser.assert_called_once_with(
@@ -340,6 +342,7 @@ async def test_empty_pipeline_is_not_scheduled(
             project_id=empty_project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
+            filtered_comp_tasks_in_db=[],
         )
     await assert_comp_runs_empty(sqlalchemy_async_engine)
     scheduler_rabbit_client_parser.assert_not_called()
@@ -355,6 +358,7 @@ async def test_empty_pipeline_is_not_scheduled(
             project_id=empty_project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
+            filtered_comp_tasks_in_db=[],
         )
     assert len(caplog.records) == 1
     assert "no computational dag defined" in caplog.records[0].message

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

@@ -169,6 +169,7 @@ async def _assert_start_pipeline(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
 
     # check the database is correctly updated, the run is published
@@ -1124,6 +1125,7 @@ async def test_broken_pipeline_configuration_is_not_scheduled_and_aborted(
         project_id=sleepers_project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
     with_disabled_scheduler_publisher.assert_called_once()
     # we shall have a a new comp_runs row with the new pipeline job
@@ -1251,6 +1253,7 @@ async def test_handling_of_disconnected_scheduler_dask(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
 
     # since there is no cluster, there is no dask-scheduler,
@@ -1766,6 +1769,7 @@ async def test_pipeline_with_on_demand_cluster_with_not_ready_backend_waits(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=True,
+        filtered_comp_tasks_in_db=[],
     )
 
     # we ask to use an on-demand cluster, therefore the tasks are published first
@@ -1870,6 +1874,7 @@ async def test_pipeline_with_on_demand_cluster_with_no_clusters_keeper_fails(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=True,
+        filtered_comp_tasks_in_db=[],
     )
 
     # we ask to use an on-demand cluster, therefore the tasks are published first

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

@@ -69,6 +69,7 @@ async def test_worker_properly_autocalls_scheduler_api(
         project_id=published_project.project.uuid,
         run_metadata=run_metadata,
         use_on_demand_clusters=False,
+        filtered_comp_tasks_in_db=[],
     )
     mocked_get_scheduler_worker.assert_called_once_with(initialized_app)
     mocked_get_scheduler_worker.return_value.apply.assert_called_once_with(
@@ -126,6 +127,7 @@ async def _project_pipeline_creation_workflow() -> None:
             project_id=published_project.project.uuid,
             run_metadata=run_metadata,
             use_on_demand_clusters=False,
+            filtered_comp_tasks_in_db=[],
         )
 
     # whatever scheduling concurrency we call in here, we shall always see the same number of calls to the scheduler